text: /*-
   * See the file LICENSE for redistribution information.
   *
   * Copyright (c) 2010, 2016 Oracle and/or its affiliates.  All rights reserved.
   *
   * $Id$
   */
  
  #include "db_config.h"
  
  #include "db_int.h"
  #include "dbinc/db_page.h"
  #include "dbinc/btree.h"
  #include "dbinc/heap.h"
  #include "dbinc/lock.h"
  #include "dbinc/mp.h"
  
  static int  __heap_bulk __P((DBC *, DBT *, u_int32_t));
  static int  __heap_getpage __P((DBC *, u_int32_t, u_int8_t *));
  static int  __heapc_close __P((DBC *, db_pgno_t, int *));
  static int  __heapc_del __P((DBC *, u_int32_t));
  static int  __heapc_destroy __P((DBC *));
  static int  __heapc_get __P((DBC *, DBT *, DBT *, u_int32_t, db_pgno_t *));
  static int  __heapc_put __P((DBC *, DBT *, DBT *, u_int32_t, db_pgno_t *));
  static int  __heapc_reloc __P((DBC *, DBT *, DBT *));
  static int  __heapc_reloc_partial __P((DBC *, DBT *, DBT *));
  static void  __heapc_search __P((DBC *, HEAPPG *, unsigned,
  		int, db_indx_t *, int *));
  static int  __heapc_split __P((DBC *, DBT *, DBT *, int));
  
  /*
   * Acquire a new page/lock.  If we are already holding a page and a lock
   * we discard those and get the new ones.  In this case we can use
   * LCK_COUPLE to save trips to lock manager.  If we are not holding a page or
   * locks, we just get a new lock and page. Lock release done with a
   * transactional lock put.
   */
  #undef  ACQUIRE
  #define	ACQUIRE(dbc, mode, lpgno, lock, fpgno, pagep, flags, mflags, ret) do { \
  	DB_MPOOLFILE *__mpf = (dbc)->dbp->mpf;				\
  	if ((pagep) != NULL) {						\
  		ret = __memp_fput(__mpf,				\
  		    (dbc)->thread_info, pagep, dbc->priority);		\
  		pagep = NULL;						\
  	}								\
  	if ((ret) == 0 && STD_LOCKING(dbc))				\
  		ret = __db_lget(dbc,					\
  		    LOCK_ISSET(lock) ? LCK_COUPLE : 0,			\
  		    lpgno, mode, flags, &(lock));			\
  	if ((ret) == 0)							\
  		ret = __memp_fget(__mpf, &(fpgno),			\
  		    (dbc)->thread_info, (dbc)->txn, mflags, &(pagep));	\
  } while (0)
  
  /* Acquire a new page/lock for a heap cursor */
  #undef  ACQUIRE_CUR
  #define	ACQUIRE_CUR(dbc, mode, p, flags, mflags, ret) do {		\
  	HEAP_CURSOR *__cp = (HEAP_CURSOR *)(dbc)->internal;		\
  	if (p != __cp->pgno)						\
  		__cp->pgno = PGNO_INVALID;				\
  	ACQUIRE(dbc, mode, p, __cp->lock, p, __cp->page, flags, mflags, ret); \
  	if ((ret) == 0) {						\
  		__cp->pgno = p;						\
  		__cp->lock_mode = (mode);				\
  	}								\
  } while (0)
  
  /* Discard the current page/lock for a cursor, indicate txn lock release */
  #undef  DISCARD
  #define	DISCARD(dbc, pagep, lock, tlock, ret) do {			\
  	DB_MPOOLFILE *__mpf = (dbc)->dbp->mpf;				\
  	int __t_ret;							\
  	__t_ret = 0;							\
  	if ((pagep) != NULL) {						\
  		__t_ret = __memp_fput(__mpf,				\
  		    (dbc)->thread_info, pagep, dbc->priority);		\
  		pagep = NULL;						\
  	}								\
  	if (__t_ret != 0 && (ret) == 0)					\
  		ret = __t_ret;						\
  	if (tlock == 1)							\
  		__t_ret = __TLPUT((dbc), lock);				\
  	else								\
  		__t_ret = __LPUT((dbc), lock);				\
  	if (__t_ret != 0 && (ret) == 0)					\
  		ret = __t_ret;						\
  } while (0)
  
  /*
   * __heapc_init --
   *	Initialize the access private portion of a cursor
   *
   * PUBLIC: int __heapc_init __P((DBC *));
   */
  int
  __heapc_init(dbc)
  	DBC *dbc;
  {
  	ENV *env;
  	int ret;
  
  	env = dbc->env;
  
  	if (dbc->internal == NULL)
  		if ((ret = __os_calloc(
  		    env, 1, sizeof(HEAP_CURSOR), &dbc->internal)) != 0)
  			return (ret);
  
  	/* Initialize methods. */
  	dbc->close = dbc->c_close = __dbc_close_pp;
  	dbc->cmp = __dbc_cmp_pp;
  	dbc->count = dbc->c_count = __dbc_count_pp;
  	dbc->del = dbc->c_del = __dbc_del_pp;
  	dbc->dup = dbc->c_dup = __dbc_dup_pp;
  	dbc->get = dbc->c_get = __dbc_get_pp;
  	dbc->pget = dbc->c_pget = __dbc_pget_pp;
  	dbc->put = dbc->c_put = __dbc_put_pp;
  	dbc->am_bulk = __heap_bulk;
  	dbc->am_close = __heapc_close;
  	dbc->am_del = __heapc_del;
  	dbc->am_destroy = __heapc_destroy;
  	dbc->am_get = __heapc_get;
  	dbc->am_put = __heapc_put;
  	dbc->am_writelock = NULL;
  
  	return (0);
  }
  
  static int
  __heap_bulk(dbc, data, flags)
  	DBC *dbc;
  	DBT *data;
  	u_int32_t flags;
  {
  	DB *dbp;
  	DB_HEAP_RID prev_rid, rid;
  	DBT sdata;
  	HEAP_CURSOR *cp;
  	HEAPBLOBHDR bhdr;
  	HEAPHDR *hdr;
  	HEAPSPLITHDR *shdr;
  	PAGE *pg;
  	db_lockmode_t lock_type;
  	int is_key, ret;
  	int32_t *offp;
  	off_t blob_size;
  	db_seq_t blob_id;
  	u_int32_t data_size, key_size, needed, space;
  	u_int8_t *dbuf, *np;
  
  	ret = 0;
  	dbp = dbc->dbp;
  	cp = (HEAP_CURSOR *)dbc->internal;
  	hdr = NULL;
  	shdr = NULL;
  
  	/* Check for additional bits for locking */
  	if (F_ISSET(dbc, DBC_RMW))
  		lock_type = DB_LOCK_WRITE;
  	else
  		lock_type = DB_LOCK_READ;
  
  	/*
  	 * np is the next place to copy things into the buffer.
  	 * dbuf always stays at the beginning of the buffer.
  	 */
  	dbuf = data->data;
  	np = dbuf;
  
  	/* Keep track of space that is left.  There is a termination entry */
  	space = data->ulen;
  	space -= sizeof(*offp);
  
  	/* Build the offset/size table from the end up. */
  	offp = (int32_t *)((u_int8_t *)dbuf + data->ulen);
  	offp--;
  
  	/*
  	 * key_size and data_size hold the 32-bit aligned size of the key and
  	 * data values written to the buffer.
  	 */
  	key_size = DB_ALIGN(DB_HEAP_RID_SZ, sizeof(u_int32_t));
  	data_size = 0;
  
  	/* is_key indicates whether keys are returned. */
  	is_key = LF_ISSET(DB_MULTIPLE_KEY) ? 1 : 0;
  
  next_pg:
  	rid.indx = cp->indx;
  	rid.pgno = cp->pgno;
  	prev_rid = rid;
  	pg = cp->page;
  
  	/*
  	 * Write records to the buffer, in the format needed by the DB_MULTIPLE
  	 * macros.  For a description of the data layout, see db.h.
  	 */
  	do {
  		if (HEAP_OFFSETTBL(dbp, pg)[rid.indx] == 0)
  			continue;
  		hdr = (HEAPHDR *)P_ENTRY(dbp, pg, rid.indx);
  		/*
  		 * If this is a split record and not the first piece of the
  		 * record, skip it.
  		 */
  		if (F_ISSET(hdr, HEAP_RECSPLIT) &&
  		    !F_ISSET(hdr, HEAP_RECFIRST))
  			continue;
  
  		/*
  		 * Calculate how much space is needed to add this record.  If
  		 * there's not enough, we're done.  If we haven't written any
  		 * data to the buffer, or if we are doing a DBP->get, return
  		 * DB_BUFFER_SMALL.
  		 */
  		needed = 0;
  		if (is_key)
  			needed = 2 * sizeof(*offp) + key_size;
  		if (F_ISSET(hdr, HEAP_RECSPLIT)) {
  			shdr = (HEAPSPLITHDR *)hdr;
  			data_size = DB_ALIGN(shdr->tsize, sizeof(u_int32_t));
  		} else if (F_ISSET(hdr, HEAP_RECBLOB)) {
  			memcpy(&bhdr, hdr, HEAPBLOBREC_SIZE);
  			GET_BLOB_SIZE(dbp->env, bhdr, blob_size, ret);
  			if (ret != 0)
  				return (ret);
  			if (blob_size > UINT32_MAX)
  				return (DB_BUFFER_SMALL);
  			data_size = (u_int32_t)blob_size;
  		} else
  			data_size = DB_ALIGN(hdr->size, sizeof(u_int32_t));
  		needed += 2 * sizeof(*offp) + data_size;
  
  		if (needed > space) {
  			if (np == dbuf || F_ISSET(dbc, DBC_FROM_DB_GET)) {
  				data->size = (u_int32_t)DB_ALIGN(
  				    needed + data->ulen - space, 1024);
  				return (DB_BUFFER_SMALL);
  			}
  			break;
  		}
  
  		if (is_key) {
  			memcpy(np, &rid, key_size);
  			*offp-- = (int32_t)(np - dbuf);
  			*offp-- = (int32_t)DB_HEAP_RID_SZ;
  			np += key_size;
  		}
  
  		if (F_ISSET(hdr, HEAP_RECSPLIT)) {
  			/*
  			 * Use __heapc_gsplit to write a split record to the
  			 * return buffer.  gsplit will return any fetched pages
  			 * to the cache, but will leave the cursor's current
  			 * page alone.
  			 */
  			memset(&sdata, 0, sizeof(DBT));
  			sdata.data = np;
  			sdata.size = sdata.ulen = shdr->tsize;
  			sdata.flags = DB_DBT_USERMEM;
  			/* gsplit expects the cursor to be positioned. */
  			cp->pgno = rid.pgno;
  			cp->indx = rid.indx;
  			if ((ret = __heapc_gsplit(
  			    dbc, &sdata, NULL, NULL)) != 0)
  				return (ret);
  		} else if (F_ISSET(hdr, HEAP_RECBLOB)) {
  			memcpy(&bhdr, hdr, HEAPBLOBREC_SIZE);
  			blob_id = (db_seq_t)bhdr.id;
  			if ((ret = __blob_bulk(
  			    dbc, data_size, blob_id, np)) != 0)
  				return (ret);
  		}else {
  			memcpy(np,
  			    (u_int8_t *)hdr + sizeof(HEAPHDR), hdr->size);
  		}
  		*offp-- = (int32_t)(np - dbuf);
  		if (F_ISSET(hdr, HEAP_RECSPLIT))
  			*offp-- = (int32_t)shdr->tsize;
  		else if (F_ISSET(hdr, HEAP_RECBLOB))
  			*offp-- = (int32_t)data_size;
  		else
  			*offp-- = (int32_t)hdr->size;
  		np += data_size;
  		space -= needed;
  		prev_rid = rid;
  
  		/*
  		 * The data and "metadata" ends of the buffer should never
  		 * overlap.
  		 */
  		DB_ASSERT(dbp->env, (void *)np <= (void *)offp);
  	} while (++rid.indx < NUM_ENT(pg));
  
  	/* If we are off the page then try the next page. */
  	if (rid.indx >= NUM_ENT(pg)) {
  		rid.pgno++;
  		ACQUIRE_CUR(dbc, lock_type, rid.pgno, 0, 0, ret);
  		if (ret == 0) {
  			cp->indx = 0;
  			goto next_pg;
  		} else if (ret != DB_PAGE_NOTFOUND)
  			return (ret);
  	}
  
  	DB_ASSERT(dbp->env, (ret == 0 || ret == DB_PAGE_NOTFOUND));
  	cp->indx = prev_rid.indx;
  	cp->pgno = prev_rid.pgno;
  
  	*offp = -1;
  
  	return (0);
  }
  
  static int
  __heapc_close(dbc, root_pgno, rmroot)
  	DBC *dbc;
  	db_pgno_t root_pgno;
  	int *rmroot;
  {
  	HEAP_CURSOR *cp;
  	int ret;
  
  	COMPQUIET(root_pgno, 0);
  	COMPQUIET(rmroot, 0);
  
  	cp = (HEAP_CURSOR *)dbc->internal;
  	ret = 0;
  
  	/* Release the page/lock held by the cursor. */
  	DISCARD(dbc, cp->page, cp->lock, 1, ret);
  	if (ret == 0 && !LOCK_ISSET(cp->lock))
  		cp->lock_mode = DB_LOCK_NG;
  
  	return (ret);
  }
  
  static int
  __heapc_del(dbc, flags)
  	DBC *dbc;
  	u_int32_t flags;
  {
  	DB *dbp;
  	DB_HEAP_RID next_rid, orig_rid;
  	DB_MPOOLFILE *mpf;
  	DBT hdr_dbt, log_dbt;
  	HEAP *h;
  	HEAPBLOBHDR bhdr;
  	HEAPHDR *hdr;
  	HEAPPG *rpage;
  	HEAP_CURSOR *cp;
  	db_pgno_t region_pgno;
  	int ret, t_ret;
  	db_seq_t blob_id;
  	u_int32_t oldspacebits, spacebits;
  	u_int16_t data_size, size;
  
  	dbp = dbc->dbp;
  	mpf = dbp->mpf;
  	h = dbp->heap_internal;
  	cp = (HEAP_CURSOR *)dbc->internal;
  	rpage = NULL;
  	ret = 0;
  	COMPQUIET(flags, 0);
  
  	/*
  	 * We need to be able to reset the cursor after deleting a record split
  	 * across multiple pages.
  	 */
  	orig_rid.pgno = cp->pgno;
  	orig_rid.indx = cp->indx;
  
  	/*
  	 * This code is always called with a page lock but no page.
  	 */
  	DB_ASSERT(dbp->env, cp->page == NULL);
  
  	/* We have a read lock, but need a write lock. */
  start:	if (STD_LOCKING(dbc) && (ret = __db_lget(dbc,
  	    LCK_COUPLE, cp->pgno, DB_LOCK_WRITE, 0, &cp->lock)) != 0)
  		return (ret);
  
  	if ((ret = __memp_fget(mpf, &cp->pgno,
  	    dbc->thread_info, dbc->txn, DB_MPOOL_DIRTY, &cp->page)) != 0)
  		return (ret);
  
  	HEAP_CALCSPACEBITS(dbp, HEAP_FREESPACE(dbp, cp->page), oldspacebits);
  
  	hdr = (HEAPHDR *)P_ENTRY(dbp, cp->page, cp->indx);
  	data_size = DB_ALIGN(hdr->size, sizeof(u_int32_t));
  	size = data_size + HEAP_HDRSIZE(hdr);
  	if (size < sizeof(HEAPSPLITHDR))
  		size = sizeof(HEAPSPLITHDR);
  	if (F_ISSET(hdr, HEAP_RECSPLIT) && !F_ISSET(hdr, HEAP_RECLAST)) {
  		next_rid.pgno = F_ISSET(hdr, HEAP_RECLAST) ?
  			PGNO_INVALID : ((HEAPSPLITHDR *)hdr)->nextpg;
  		next_rid.indx = F_ISSET(hdr, HEAP_RECLAST) ?
  			PGNO_INVALID : ((HEAPSPLITHDR *)hdr)->nextindx;
  	} else {
  		next_rid.pgno = PGNO_INVALID;
  		next_rid.indx = 0;
  	}
  
  	/* Delete the blob file. */
  	if (F_ISSET(hdr, HEAP_RECBLOB)) {
  		memcpy(&bhdr, hdr, HEAPBLOBREC_SIZE);
  		blob_id = (db_seq_t)bhdr.id;
  		if ((ret = __blob_del(dbc, blob_id)) != 0)
  			return (ret);
  	}
  
  	/* Log the deletion. */
  	if (DBC_LOGGING(dbc)) {
  		hdr_dbt.data = hdr;
  		hdr_dbt.size = HEAP_HDRSIZE(hdr);
  		log_dbt.data = (u_int8_t *)hdr + hdr_dbt.size;
  		log_dbt.size = data_size;
  		if ((ret = __heap_addrem_log(dbp, dbc->txn, &LSN(cp->page),
  		    0, OP_SET(DB_REM_HEAP, cp->page),
  		    cp->pgno, (u_int32_t)cp->indx, size,
  		    &hdr_dbt, &log_dbt, &LSN(cp->page))) != 0)
  			goto err;
  	} else
  		LSN_NOT_LOGGED(LSN(cp->page));
  
  	if ((ret = __heap_ditem(dbc, cp->page, cp->indx, size)) != 0)
  		goto err;
  
  	/*
  	 * If the deleted item lived in a region prior to our current, back up
  	 * the current region, giving us a chance to reuse the newly available
  	 * space on the next insert.
  	 */
  	region_pgno = HEAP_REGION_PGNO(dbp, cp->pgno);
  	if (region_pgno < h->curregion)
  		h->curregion = region_pgno;
  
  	HEAP_CALCSPACEBITS(dbp, HEAP_FREESPACE(dbp, cp->page), spacebits);
  
  	if (spacebits != oldspacebits) {
  		/*
  		 * Get the region page.  We never lock the region page, the data
  		 * page lock locks the corresponding bits in the bitmap and
  		 * latching serializes access.
  		 */
  		if ((ret = __memp_fget(mpf, &region_pgno,
  		    dbc->thread_info, NULL, DB_MPOOL_DIRTY, &rpage)) != 0)
  			goto err;
  		HEAP_SETSPACE(dbp, rpage,
  		    (cp->pgno - region_pgno) - 1, spacebits);
  	}
  
  err:	DB_ASSERT(dbp->env, ret != DB_PAGE_NOTFOUND);
  	if (rpage != NULL && (t_ret = __memp_fput(mpf,
  	    dbc->thread_info, rpage, dbc->priority)) != 0 && ret == 0)
  		ret = t_ret;
  	rpage = NULL;
  
  	if ((t_ret = __memp_fput(mpf,
  	    dbc->thread_info, cp->page, dbc->priority)) != 0 && ret == 0)
  		ret = t_ret;
  
  	cp->page = NULL;
  
  	if (ret == 0 && next_rid.pgno != PGNO_INVALID) {
  		cp->pgno = next_rid.pgno;
  		cp->indx = next_rid.indx;
  		goto start;
  	}
  
  	cp->pgno = orig_rid.pgno;
  	cp->indx = orig_rid.indx;
  
  	return (ret);
  }
  
  /*
   * __heap_ditem --
   *   Remove an item from a page.  Note when deleting blob records that the file
   *   has to be deleted separate from calling this function.
   *
   * PUBLIC: int __heap_ditem
   * PUBLIC:   __P((DBC *, PAGE *, u_int32_t, u_int32_t));
   */
  int
  __heap_ditem(dbc, pagep, indx, nbytes)
  	DBC *dbc;
  	PAGE *pagep;
  	u_int32_t indx, nbytes;
  {
  	DB *dbp;
  	db_indx_t first, i, max, off, *offtbl, span;
  	u_int8_t *src, *dest;
  
  	dbp = dbc->dbp;
  
  	DB_ASSERT(dbp->env, TYPE(pagep) == P_HEAP);
  	DB_ASSERT(dbp->env, nbytes == DB_ALIGN(nbytes, sizeof(u_int32_t)));
  	DB_ASSERT(dbp->env, nbytes >= sizeof(HEAPSPLITHDR));
  
  	offtbl = (db_indx_t *)HEAP_OFFSETTBL(dbp, pagep);
  	off = offtbl[indx];
  	/*
  	 * Find the lowest offset on the page, and adjust offsets that are about
  	 * to be moved.  If the deleted item is the lowest offset on the page,
  
  	 * everything will work, that is not a special case.
  	 */
  	max = HEAP_HIGHINDX(pagep);
  	first = HOFFSET(pagep);
  	for (i = 0; i <= max; i++) {
  		if (offtbl[i] < off && offtbl[i] != 0)
  			offtbl[i] += nbytes;
  	}
  	offtbl[indx] = 0;
  
  	/*
  	 * Coalesce free space at the beginning of the page.  Shift all the data
  	 * preceding the deleted entry down, overwriting the deleted entry.
  	 */
  	src = (u_int8_t *)(pagep) + first;
  	dest = src + nbytes;
  	span = off - first;
  	memmove(dest, src, span);
  #ifdef DIAGNOSTIC
  	memset(src, CLEAR_BYTE, nbytes);
  #endif
  
  	/* Update the page's metadata. */
  	NUM_ENT(pagep)--;
  	HOFFSET(pagep) += nbytes;
  	if (indx < HEAP_FREEINDX(pagep))
  		HEAP_FREEINDX(pagep) = indx;
  	while (HEAP_HIGHINDX(pagep) > 0 && offtbl[HEAP_HIGHINDX(pagep)] == 0)
  		HEAP_HIGHINDX(pagep)--;
  	if (NUM_ENT(pagep) == 0)
  		HEAP_FREEINDX(pagep) = 0;
  	else if (HEAP_FREEINDX(pagep) > HEAP_HIGHINDX(pagep) + 1)
  		HEAP_FREEINDX(pagep) = HEAP_HIGHINDX(pagep) + 1;
  
  	return (0);
  }
  
  static int
  __heapc_destroy(dbc)
  	DBC *dbc;
  {
  	HEAP_CURSOR *cp;
  
  	cp = (HEAP_CURSOR *)dbc->internal;
  	__os_free(dbc->env, cp);
  	dbc->internal = NULL;
  
  	return (0);
  }
  
  /*
   * __heapc_get --
   *	Get using a cursor (heap).
   */
  static int
  __heapc_get(dbc, key, data, flags, pgnop)
  	DBC *dbc;
  	DBT *key;
  	DBT *data;
  	u_int32_t flags;
  	db_pgno_t *pgnop;
  {
  	DB *dbp;
  	DB_HEAP_RID rid;
  	DB_MPOOLFILE *mpf;
  	DB_LOCK meta_lock;
  	DBT tmp_val;
  	HEAPBLOBHDR bhdr;
  	HEAPHDR *hdr;
  	HEAPMETA *meta;
  	HEAPPG *dpage;
  	HEAP_CURSOR *cp;
  	db_lockmode_t lock_type;
  	db_pgno_t pgno;
  	int cmp, np_inc, f_indx, found, getpage, indx, ret;
  	off_t blob_size;
  	db_seq_t blob_id;
  
  	dbp = dbc->dbp;
  	mpf = dbp->mpf;
  	cp = (HEAP_CURSOR *)dbc->internal;
  	pgno = PGNO_INVALID;
  	LOCK_INIT(meta_lock);
  	COMPQUIET(pgnop, NULL);
  
  	if (F_ISSET(key, DB_DBT_USERMEM) && key->ulen < DB_HEAP_RID_SZ) {
  		key->size = DB_HEAP_RID_SZ;
  		return (DB_BUFFER_SMALL);
  	}
  
  	/* Check for additional bits for locking */
  	if (F_ISSET(dbc, DBC_RMW))
  		lock_type = DB_LOCK_WRITE;
  	else
  		lock_type = DB_LOCK_READ;
  
  	np_inc = ret = 0;
  	found = getpage = FALSE;
  	meta = NULL;
  	dpage = NULL;
  	switch (flags) {
  	case DB_CURRENT:
  
  		/*
  		 * Acquire the current page with read lock unless user
  		 * has asked for a write lock.  Ensure page and record
  		 * exist still.
  		 */
  		ACQUIRE_CUR(dbc, lock_type, cp->pgno, 0, 0, ret);
  		if (ret != 0) {
  			if (ret == DB_PAGE_NOTFOUND)
  				ret = DBC_ERR(dbc, DB_NOTFOUND);
  			goto err;
  		}
  
  		if (HEAP_OFFSETTBL(dbp, cp->page)[cp->indx] == 0) {
  			ret = DB_NOTFOUND;
  			goto err;
  		}
  		dpage = (HEAPPG *)cp->page;
  		hdr = (HEAPHDR *)P_ENTRY(dbp, dpage, cp->indx);
  		if (F_ISSET(hdr, HEAP_RECSPLIT) &&
  		    !F_ISSET(hdr, HEAP_RECFIRST)) {
  			ret = DBC_ERR(dbc, DB_NOTFOUND);
  			goto err;
  		}
  
  		break;
  	case DB_FIRST:
  		/*
  		 * The region pages do not distinguish between an empty
  		 * page and page with a something on it.  So, we will
  		 * grab the first possible data page and look for the
  		 * lowest index with data.  If page is empty we go on to
  		 * the next page and look. If no page, then no records.
  		 */
  first:		pgno = FIRST_HEAP_DPAGE;
  		while (!found) {
  			/* Put old lock/page and get the new lock/page */
  			ACQUIRE_CUR(dbc, lock_type, pgno, 0, 0, ret);
  			if (ret != 0 ) {
  				if (ret == DB_PAGE_NOTFOUND)
  					ret = DBC_ERR(dbc, DB_NOTFOUND);
  				goto err;
  			}
  			dpage = (HEAPPG *)cp->page;
  			/*
  			 * The page needs to be a data page with entries on
  			 * it.  If page is good, loop through the offset table
  			 * finding first non-split record or first piece of a
  			 * split record, then set up cursor.
  			 */
  			__heapc_search(dbc, dpage, 0, 1, &cp->indx, &found);
  			if (found)
  				cp->pgno = pgno;
  			else
  				pgno++;
  		}
  		break;
  	case DB_LAST:
  		/*
  		 * Grab the metadata page to find the last page, and start
  		 * there looking backwards for the record with the highest
  		 * index and return that one.
  		 */
  last:		pgno = PGNO_BASE_MD;
  		ACQUIRE(dbc, DB_LOCK_READ,
  		    pgno, meta_lock, pgno, meta, 0, 0, ret);
  		if (ret != 0)
  			goto err;
  
  		pgno = meta->dbmeta.last_pgno;
  
  		/*
  		 * It is possible to have another page added while we are
  		 * searching backwards for last record. No need to block
  		 * this case from occurring by keeping meta page lock.
  		 */
  		DISCARD(dbc, meta, meta_lock, 1, ret);
  		if (ret != 0)
  			goto err;
  
  		while (!found) {
  			/* Don't look earlier than the first data page. */
  			if (pgno < FIRST_HEAP_DPAGE) {
  				ret = DBC_ERR(dbc, DB_NOTFOUND);
  				goto err;
  			}
  
  			/* Put old lock/page and get the new lock/page. */
  			ACQUIRE_CUR(dbc, lock_type, pgno, 0, 0, ret);
  			if (ret != 0)
  				goto err;
  			dpage = (HEAPPG *)cp->page;
  			/*
  			 * The page needs to be a data page with entries on
  			 * it.  If page is good, search backwards until the a
  			 * non-split record or the first piece of a split record
  			 * is found.
  			 */
  			__heapc_search(dbc,
  			    dpage, HEAP_HIGHINDX(dpage), 1, &cp->indx, &found);
  			if (found)
  				cp->pgno = pgno;
  			else
  				pgno--;
  		}
  		break;
  	case DB_NEXT_NODUP:
  	case DB_NEXT:
  	case DB_PREV_NODUP:
  	case DB_PREV:
  		/*
  		 * np_inc stores whether to increment or decrement when
  		 * iterating through records on a page and pages in the file.
  		 */
  		if (flags == DB_NEXT_NODUP || flags == DB_NEXT)
  			np_inc = 1;
  		else
  			np_inc = -1;
  		/* If cursor not initialized, behave as DB_FIRST/DB_LAST */
  		if (dbc->internal->pgno == PGNO_INVALID) {
  			if (np_inc == 1)
  				goto first;
  			else
  				goto last;
  		}
  
  		/*
  		 * Acquire the current page with the lock we have already,
  		 * unless user has asked for a write lock.
  		 */
  		ACQUIRE_CUR(dbc, lock_type, cp->pgno, 0, 0, ret);
  		if (ret != 0)
  			goto err;
  		dpage = (HEAPPG *)cp->page;
  
  		if (np_inc == 1 && cp->indx >= HEAP_HIGHINDX(dpage))
  			/* At end of current page, must get next page. */
  			getpage = TRUE;
  		else if (np_inc == -1) {
  			/*
  			 * Loop through indexes and find first used slot.  Check
  			 * if already at the first slot.
  			 */
  			for (f_indx=0; (f_indx <= HEAP_HIGHINDX(dpage)) &&
  			    (HEAP_OFFSETTBL(dbp, dpage)[f_indx] == 0); f_indx++)
  			{
  				/* No-op. */
  			}
  
  			/* At the beginning of current page, get new page */
  			if (cp->indx == 0 || cp->indx <= f_indx) {
  				if (cp->pgno == FIRST_HEAP_DPAGE) {
  					ret = DBC_ERR(dbc, DB_NOTFOUND);
  					goto err;
  				}
  				getpage = TRUE;
  			}
  		}
  
  		while (!found) {
  			if (getpage) {
  				if (np_inc == -1)
  					pgno = cp->pgno - 1;
  				else if (np_inc == 1)
  					pgno = cp->pgno + 1;
  				if (pgno < FIRST_HEAP_DPAGE) {
  					ret = DBC_ERR(dbc, DB_NOTFOUND);
  					goto err;
  				}
  				/* Put current page/lock and get next one */
  				ACQUIRE_CUR(dbc, lock_type, pgno, 0, 0, ret);
  				if (ret != 0) {
  					if (np_inc == 1 &&
  					    ret == DB_PAGE_NOTFOUND)
  						/* Beyond last page */
  						ret = DBC_ERR(dbc, DB_NOTFOUND);
  					goto err;
  				}
  				dpage = (HEAPPG *)cp->page;
  
  				/*
  				 * If page is a spam page or its a data
  				 * page without entries, try again.
  				 */
  				if (TYPE(dpage) != P_HEAP ||
  				    (TYPE(dpage) == P_HEAP &&
  				    NUM_ENT(dpage) == 0))
  					continue;
  
  				if (np_inc == 1)
  					/*
  					 * When searching, indx gets
  					 * bumped to 0
  					 */
  					cp->indx = UINT16_MAX;
  				else
  					/*
  					 * When searching, indx gets bumped to
  					 * high indx
  					 */
  					cp->indx = HEAP_HIGHINDX(dpage) + 1;
  				getpage = FALSE;
  			}
  
  			/*
  			 * Bump index and loop through the offset table finding
  			 * first nonzero entry.  If the offset is for a split
  			 * record, make sure it's the first piece of the split
  			 * record. HEAP_HIGHINDX always points to highest filled
  			 * entry on page.
  			 */
  			if (np_inc == -1)
  				cp->indx--;
  			else if (np_inc == 1)
  				cp->indx++;
  			__heapc_search(dbc,
  			    dpage, cp->indx, np_inc, &cp->indx, &found);
  			if (!found)
  				/* Nothing of interest on page, so try next */
  				getpage = TRUE;
  		}
  		break;
  	case DB_GET_BOTH_RANGE:
  	case DB_GET_BOTH:
  	case DB_SET_RANGE:
  	case DB_SET:
  		pgno = ((DB_HEAP_RID *)key->data)->pgno;
  		indx = ((DB_HEAP_RID *)key->data)->indx;
  
  		/* First make sure we're trying to get a data page. */
  		if (pgno == PGNO_BASE_MD ||
  		    pgno == HEAP_REGION_PGNO(dbp, pgno)) {
  			ret = DBC_ERR(dbc, DB_NOTFOUND);
  			goto err;
  		}
  
  		/* Lock the data page and get it. */
  		ACQUIRE_CUR(dbc, lock_type, pgno, 0, 0, ret);
  
  		if (ret != 0) {
  			if (ret == DB_PAGE_NOTFOUND)
  				ret = DBC_ERR(dbc, DB_NOTFOUND);
  			goto err;
  		}
  		dpage = (HEAPPG *)cp->page;
  
  		/* validate requested index, throw error if not in range */
  		if ((indx >  HEAP_HIGHINDX(dpage)) ||
  		    (HEAP_OFFSETTBL(dbp, dpage)[indx] == 0)) {
  			DISCARD(dbc, cp->page, cp->lock, 0, ret);
  			ret = DBC_ERR(dbc, DB_NOTFOUND);
  			goto err;
  		}
  		hdr = (HEAPHDR *)P_ENTRY(dbp, dpage, indx);
  		if (F_ISSET(hdr, HEAP_RECSPLIT) &&
  		    !F_ISSET(hdr, HEAP_RECFIRST)) {
  			DISCARD(dbc, cp->page, cp->lock, 0, ret);
  			ret = DBC_ERR(dbc, DB_NOTFOUND);
  			goto err;
  		}
  
  		cp->pgno = pgno;
  		cp->indx = indx;
  
  		if (flags == DB_GET_BOTH || flags == DB_GET_BOTH_RANGE)	{
  			memset(&tmp_val, 0, sizeof(DBT));
  			/* does the data match ? */
  			if (F_ISSET(hdr, HEAP_RECSPLIT)) {
  				tmp_val.flags = DB_DBT_MALLOC;
  				if ((ret = __heapc_gsplit(
  				    dbc, &tmp_val, NULL, 0)) != 0)
  					goto err;
  			} else if (F_ISSET(hdr, HEAP_RECBLOB)) {
  				memcpy(&bhdr, hdr, HEAPBLOBREC_SIZE);
  				blob_id = (db_seq_t)bhdr.id;
  				GET_BLOB_SIZE(dbc->env, bhdr, blob_size, ret);
  				if (ret != 0)
  					goto err;
  				if (blob_size > UINT32_MAX) {
  					ret = DB_BUFFER_SMALL;
  					goto err;
  				}
  				tmp_val.flags = DB_DBT_MALLOC;
  				if ((ret = __blob_get(dbc, &tmp_val,
  				    blob_id, blob_size, NULL, 0)) != 0)
  					goto err;
  			} else {
  				tmp_val.data =
  				    (void *)((u_int8_t *)hdr + sizeof(HEAPHDR));
  				tmp_val.size = hdr->size;
  			}
  			cmp = __dbt_defcmp(dbp, &tmp_val, data, NULL);
  			if (F_ISSET(&tmp_val, DB_DBT_MALLOC))
  				__os_ufree(dbp->env, tmp_val.data);
  			if (cmp != 0) {
  				ret = DBC_ERR(dbc, DB_NOTFOUND);
  				goto err;
  			}
  		}
  
  		break;
  	case DB_NEXT_DUP:
  	case DB_PREV_DUP:
  		ret = DBC_ERR(dbc, DB_NOTFOUND);
  		goto err;
  	default:
  		/* DB_GET_RECNO, DB_JOIN_ITEM, DB_SET_RECNO are invalid */
  		ret = __db_unknown_flag(dbp->env, "__heap_get", flags);
  		goto err;
  
  	}
  
  err:	if (ret == 0 ) {
  		if (key != NULL) {
  			rid.pgno = cp->pgno;
  			rid.indx = cp->indx;
  			ret = __db_retcopy(dbp->env, key, &rid,
  			    DB_HEAP_RID_SZ, &dbc->rkey->data, &dbc->rkey->ulen);
  			F_SET(key, DB_DBT_ISSET);
  		}
  
  	} else {
  		if (meta != NULL)
  			(void)__memp_fput(mpf,
  			    dbc->thread_info, meta, dbc->priority);
  		if (LOCK_ISSET(meta_lock))
  			(void)__LPUT(dbc, meta_lock);
  		if (LOCK_ISSET(cp->lock))
  			 (void)__LPUT(dbc, cp->lock);
  	}
  	DB_ASSERT(dbp->env, ret != DB_PAGE_NOTFOUND);
  	return (ret);
  }
  
  /*
   * __heapc_search --
   *	Search a given a heap page, starting at a given index, for a viable heap
   *	record.  Return the index of the found record in indxp.
   *
   * Compiler Note:
   *	The 'begin' parameter would be a db_indx_t except that unsigned K&R C
   *	function parameters that are smaller than a full word are promoted to
   *	signed integers. By declaring 'begin' as a natural unsigned, we avoid
   *	a warning generated by gcc -Wall:
   *		promoted type 'int' of K&R function parameter is not compatible
   *		with the parameter type 'db_indx_t' ...
   */
  static void
  __heapc_search(dbc, dpage, begin, dir, indxp, found)
       DBC *dbc;
       HEAPPG *dpage;
       unsigned begin;
       int dir;
       db_indx_t *indxp;
       int *found;
  {
  	DB *dbp;
  	HEAPHDR *hdr;
  	db_indx_t indx;
  
  	dbp = dbc->dbp;
  	DB_ASSERT(dbp->env, dir == -1 || dir == 1);
  
  	*found = FALSE;
  	if (TYPE(dpage) != P_HEAP || NUM_ENT(dpage) == 0)
  		return;
  
  	indx = (db_indx_t) begin;
  	for (;;) {
  		if (HEAP_OFFSETTBL(dbp, dpage)[indx] != 0) {
  			hdr = (HEAPHDR *)P_ENTRY(dbp, dpage, indx);
  			if (!F_ISSET(hdr, HEAP_RECSPLIT) ||
  			    F_ISSET(hdr, HEAP_RECFIRST)) {
  				*found = TRUE;
  				*indxp = indx;
  				break;
  			}
  		}
  		if ((dir == -1 && indx == 0) ||
  		    (dir == 1 && indx == HEAP_HIGHINDX(dpage)))
  			break;
  
  		if (dir == -1)
  			indx--;
  		else
  			indx++;
  	}
  }
  
  #undef	IS_FIRST
  #define	IS_FIRST (last_rid.pgno == PGNO_INVALID)
  /*
   * __heapc_reloc_partial --
   *	 Move data from a too-full page to a new page.  The old data page must
   *	 be write locked before calling this method.
   */
  static int
  __heapc_reloc_partial(dbc, key, data)
  	DBC *dbc;
  	DBT *key;
  	DBT *data;
  {
  	DB *dbp;
  	DBT hdr_dbt, log_dbt, t_data, t_key;
  	DB_HEAP_RID last_rid, next_rid;
  	HEAPHDR *old_hdr;
  	HEAPSPLITHDR new_hdr;
  	HEAP_CURSOR *cp;
  	int add_bytes, ret;
  	u_int32_t buflen, data_size, dlen, doff, left, old_size;
  	u_int32_t remaining, size;
  	u_int8_t *buf, *olddata;
  
  	dbp = dbc->dbp;
  	cp = (HEAP_CURSOR *)dbc->internal;
  	old_hdr = (HEAPHDR *)(P_ENTRY(dbp, cp->page, cp->indx));
  	memset(&hdr_dbt, 0, sizeof(DBT));
  	memset(&log_dbt, 0, sizeof(DBT));
  	buf = NULL;
  	COMPQUIET(key, NULL);
  
  	/* We only work on partial puts. */
  	DB_ASSERT(dbp->env, F_ISSET(data, DB_DBT_PARTIAL));
  	DB_ASSERT(dbp->env, !F_ISSET(old_hdr, HEAP_RECBLOB));
  
  	/*
  	 * Start by calculating the data_size, total size of the new record, and
  	 * dlen, the number of bytes we will actually overwrite.  Keep a local
  	 * copy of doff, we'll adjust it as we see pieces of the record so that
  	 * it's always relative to the current piece of data.
  	 */
  	if (F_ISSET(old_hdr, HEAP_RECSPLIT))
  		old_size = ((HEAPSPLITHDR *)old_hdr)->tsize;
  	else
  		old_size = old_hdr->size;
  	doff = data->doff;
  	if (old_size < doff) {
  		/* Post-pending */
  		dlen = data->dlen;
  		data_size = doff + data->size;
  	} else {
  		if (old_size - doff < data->dlen)
  			dlen = old_size - doff;
  		else
  			dlen = data->dlen;
  		data_size = (old_size - dlen) + data->size;
  	}
  
  	/*
  	 * We don't need a buffer large enough to hold the data_size
  	 * bytes, just one large enough to hold the bytes that will be
  	 * written to an individual page.  We'll realloc to the necessary size
  	 * as needed.
  	 */
  	buflen = 0;
  	buf = NULL;
  
  	/*
  	 * We are updating an existing record, which will grow into a split
  	 * record.  The strategy is to overwrite the existing record (or each
  	 * piece of the record if the record is already split.)  If the new
  	 * record is shorter than the old, delete any extra pieces.  If the new
  	 * record is longer than the old, use heapc_split() to write the extra
  	 * data.
  	 *
  	 * We start each loop with old_hdr pointed at the header for the old
  	 * record and the necessary page write locked in cp->page.
  	 */
  	last_rid.pgno = PGNO_INVALID;
  	last_rid.indx = 0;
  	add_bytes = 1;
  	left = data_size;
  	memset(&t_data, 0, sizeof(DBT));
  	remaining = 0;
  	for (;;) {
  		/* Figure out if we have a next piece. */
  		if (F_ISSET(old_hdr, HEAP_RECSPLIT)) {
  			next_rid.pgno = ((HEAPSPLITHDR *)old_hdr)->nextpg;
  			next_rid.indx = ((HEAPSPLITHDR *)old_hdr)->nextindx;
  		} else {
  			next_rid.pgno = PGNO_INVALID;
  			next_rid.indx = 0;
  		}
  
  		/*
  		 * Before we delete the old data, use it to construct the new
  		 * data. First figure out the size of the new piece, including
  		 * any remaining data from the last piece.
  		 */
  		if (doff >= old_hdr->size)
  			if (F_ISSET(old_hdr, HEAP_RECLAST) ||
  			    !F_ISSET(old_hdr, HEAP_RECSPLIT)) {
  				/* Post-pending. */
  				data_size = doff + data->size;
  			} else {
  				/* The new piece is just the old piece. */
  				data_size = old_hdr->size;
  			}
  		else if (doff + dlen > old_hdr->size)
  			/*
  			 * Some of the to-be-overwritten bytes are on the next
  			 * piece, but we'll append all the new bytes to this
  			 * piece if we haven't already written them.
  			 */
  			data_size = doff + (add_bytes ? data->size : 0);
  		else
  			data_size = (old_hdr->size -
  				dlen) + (add_bytes ? data->size : 0);
  		data_size += remaining;
  
  		if (data_size > buflen) {
  			if (__os_realloc(dbp->env, data_size, &buf) != 0)
  				return (ENOMEM);
  			buflen = data_size;
  		}
  		t_data.data = buf;
  
  		/*
  		 * Adjust past any remaining bytes, they've already been moved
  		 * to the beginning of the buffer.
  		 */
  		buf += remaining;
  		remaining = 0;
  
  		olddata = (u_int8_t *)old_hdr + HEAP_HDRSIZE(old_hdr);
  		if (doff >= old_hdr->size) {
  			memcpy(buf, olddata, old_hdr->size);
  			doff -= old_hdr->size;
  			if (F_ISSET(old_hdr, HEAP_RECLAST) ||
  			    !F_ISSET(old_hdr, HEAP_RECSPLIT)) {
  				/* Post-pending. */
  				buf += old_hdr->size;
  				memset(buf, '\0', doff);
  				buf += doff;
  				memcpy(buf, data->data, data->size);
  			}
  		} else {
  			/* Preserve the first doff bytes. */
  			memcpy(buf, olddata, doff);
  			buf += doff;
  			olddata += doff;
  			/* Copy in the new bytes, if needed. */
  			if (add_bytes) {
  				memcpy(buf, data->data, data->size);
  				buf += data->size;
  				add_bytes = 0;
  			}
  			/* Skip dlen bytes. */
  			if (doff + dlen < old_hdr->size) {
  				olddata += dlen;
  				memcpy(buf,
  				    olddata, (old_hdr->size - doff) - dlen);
  				dlen = 0;
  			} else
  				/*
  				 * The data to be removed spills over onto the
  				 * following page(s).  Adjust dlen to account
  				 * for the bytes removed from this page.
  				 */
  				dlen = doff + dlen - old_hdr->size;
  			doff = 0;
  		}
  		buf = t_data.data;
  
  		/* Delete the old data, after logging it. */
  		old_size = DB_ALIGN(
  		    old_hdr->size + HEAP_HDRSIZE(old_hdr), sizeof(u_int32_t));
  		if (old_size < sizeof(HEAPSPLITHDR))
  			old_size = sizeof(HEAPSPLITHDR);
  		if (DBC_LOGGING(dbc)) {
  			hdr_dbt.data = old_hdr;
  			hdr_dbt.size = HEAP_HDRSIZE(old_hdr);
  			log_dbt.data = (u_int8_t *)old_hdr + hdr_dbt.size;
  			log_dbt.size = DB_ALIGN(
  			    old_hdr->size, sizeof(u_int32_t));
  			if ((ret = __heap_addrem_log(dbp, dbc->txn,
  			    &LSN(cp->page), 0, OP_SET(DB_REM_HEAP, cp->page),
  			    cp->pgno, (u_int32_t)cp->indx, old_size,
  			    &hdr_dbt, &log_dbt, &LSN(cp->page))) != 0)
  				goto err;
  		} else
  			LSN_NOT_LOGGED(LSN(cp->page));
  		if ((ret = __heap_ditem(
  		    dbc, cp->page, cp->indx, old_size)) != 0)
  			goto err;
  
  		if (left == 0)
  			/*
  			 * We've finished writing the new record, we're just
  			 * cleaning up the old record now.
  			 */
  			goto next_pg;
  
  		if (data_size == 0 && !IS_FIRST) {
  			/*
  			 * This piece is being completely removed.  We need to
  			 * adjust the header of the previous piece now.
  			 */
  			ACQUIRE_CUR(dbc, DB_LOCK_WRITE,
  			    last_rid.pgno, 0, DB_MPOOL_DIRTY, ret);
  			if (ret != 0)
  				goto err;
  
  			cp->indx = last_rid.indx;
  			old_hdr = (HEAPHDR *)(P_ENTRY(dbp, cp->page, cp->indx));
  
  			if (DBC_LOGGING(dbc)) {
  				old_size = DB_ALIGN(old_hdr->size +
  				    HEAP_HDRSIZE(old_hdr), sizeof(u_int32_t));
  				hdr_dbt.data = old_hdr;
  				hdr_dbt.size = HEAP_HDRSIZE(old_hdr);
  				log_dbt.data =
  				    (u_int8_t *)old_hdr + hdr_dbt.size;
  				log_dbt.size = DB_ALIGN(
  				    old_hdr->size, sizeof(u_int32_t));
  				if ((ret = __heap_addrem_log(dbp, dbc->txn,
  				    &LSN(cp->page), 0,
  				    OP_SET(DB_REM_HEAP, cp->page), cp->pgno,
  				    (u_int32_t)cp->indx, old_size,
  				    &hdr_dbt, &log_dbt, &LSN(cp->page))) != 0)
  					goto err;
  			} else
  				LSN_NOT_LOGGED(LSN(cp->page));
  
  			((HEAPSPLITHDR *)old_hdr)->nextpg = next_rid.pgno;
  			((HEAPSPLITHDR *)old_hdr)->nextindx = next_rid.indx;
  
  			if (DBC_LOGGING(dbc)) {
  				if ((ret = __heap_addrem_log(dbp, dbc->txn,
  				    &LSN(cp->page), 0,
  				    OP_SET(DB_ADD_HEAP, cp->page), cp->pgno,
  				    (u_int32_t)cp->indx, old_size,
  				    &hdr_dbt, &log_dbt, &LSN(cp->page))) != 0)
  					goto err;
  			} else
  				LSN_NOT_LOGGED(LSN(cp->page));
  
  			DISCARD(dbc, cp->page, cp->lock, 1, ret);
  
  			goto next_pg;
  		}
  
  		/* Set up the header for the new record. */
  		memset(&new_hdr, 0, sizeof(HEAPSPLITHDR));
  		new_hdr.std_hdr.flags = HEAP_RECSPLIT;
  		/*
  		 * If next_rid.pgno == PGNO_INVALID and there's still more data,
  		 * we'll come back and correct the header once we know where the
  		 * next piece lives.
  		 */
  		new_hdr.nextpg = next_rid.pgno;
  		new_hdr.nextindx = next_rid.indx;
  		/*
  		 * Figure out how much we can fit on the page, rounding down to
  		 * a multiple of 4.  If we will have to expand the offset table,
  		 * account for that. It needs to be enough to at least fit the
  		 * split header.
  		 */
  		size = HEAP_FREESPACE(dbp, cp->page);
  		if (NUM_ENT(cp->page) == 0 ||
  		    cp->indx > HEAP_HIGHINDX(cp->page))
  			size -= sizeof(db_indx_t);
  		/* Round down to a multiple of 4. */
  		size = DB_ALIGN(
  		    (size - sizeof(u_int32_t)) + 1, sizeof(u_int32_t));
  		DB_ASSERT(dbp->env, size >= sizeof(HEAPSPLITHDR));
  
  		/*
  		 * We try to fill the page, but cannot write more than
  		 * t_data.size bytes, that's all we have in-memory.
  		 */
  		new_hdr.std_hdr.size = (u_int16_t)
  		    (size - sizeof(HEAPSPLITHDR));
  		if (new_hdr.std_hdr.size > data_size)
  			new_hdr.std_hdr.size = data_size;
  		if (new_hdr.std_hdr.size >= left) {
  			new_hdr.std_hdr.size = left;
  			new_hdr.std_hdr.flags |= HEAP_RECLAST;
  			new_hdr.nextpg = PGNO_INVALID;
  			new_hdr.nextindx = 0;
  		}
  		if (IS_FIRST) {
  			new_hdr.std_hdr.flags |= HEAP_RECFIRST;
  			new_hdr.tsize = left;
  		}
  
  		/* Now write the new data to the page. */
  		t_data.size = new_hdr.std_hdr.size;
  		hdr_dbt.data = &new_hdr;
  		hdr_dbt.size = sizeof(HEAPSPLITHDR);
  		/* Log the write. */
  		if (DBC_LOGGING(dbc)) {
  			if ((ret = __heap_addrem_log(dbp,
  			    dbc->txn, &LSN(cp->page), 0,
  			    OP_SET(DB_ADD_HEAP, cp->page), cp->pgno,
  			    (u_int32_t)cp->indx,
  			    size, &hdr_dbt, &t_data, &LSN(cp->page))) != 0)
  				goto err;
  		} else
  			LSN_NOT_LOGGED(LSN(cp->page));
  		if ((ret = __heap_pitem(dbc,
  		    (PAGE *)cp->page, cp->indx, size, &hdr_dbt, &t_data)) != 0)
  			goto err;
  
  		left -= new_hdr.std_hdr.size;
  		/*
  		 * If any data couldn't fit on this page, it has to go onto the
  		 * next.  Copy it to the front of the buffer and it will be
  		 * preserved in the next loop.
  		 */
  		if (new_hdr.std_hdr.size < data_size) {
  			remaining = data_size - new_hdr.std_hdr.size;
  			memmove(buf, buf + new_hdr.std_hdr.size, remaining);
  		}
  
  		/*
  		 * Remember this piece's RID, we may need to update the header
  		 * if the next data piece is removed, or if this is the final
  		 * piece and we add data to the end of the record.
  		 */
  next_pg:	last_rid.pgno = cp->pgno;
  		last_rid.indx = cp->indx;
  		/* Get the next page, if any. */
  		if (next_rid.pgno != PGNO_INVALID) {
  			ACQUIRE_CUR(dbc, DB_LOCK_WRITE,
  			    next_rid.pgno, 0, DB_MPOOL_DIRTY, ret);
  			if (ret != 0)
  				goto err;
  			cp->indx = next_rid.indx;
  			old_hdr = (HEAPHDR *)(P_ENTRY(dbp, cp->page, cp->indx));
  			DB_ASSERT(dbp->env,
  			    HEAP_HIGHINDX(cp->page) <= cp->indx);
  			DB_ASSERT(dbp->env, F_ISSET(old_hdr, HEAP_RECSPLIT));
  		} else {
  			/* Discard the page and drop the lock, txn-ally. */
  			DISCARD(dbc, cp->page, cp->lock, 1, ret);
  			if (ret != 0)
  				goto err;
  			break;
  		}
  	}
  
  	/*
  	 * If there is more work to do, let heapc_split do it.  After
  	 * heapc_split returns we need to update nextpg and nextindx in the
  	 * header of the last piece we wrote above.
  	 *
  	 * For logging purposes, we "delete" the old record and then "add" the
  	 * record.  This makes redo/undo work as-is, but we won't actually
  	 * delete and re-add the record.
  	 */
  	if (left > 0) {
  		memset(&t_key, 0, sizeof(DBT));
  		t_key.size = t_key.ulen = sizeof(DB_HEAP_RID);
  		t_key.data = &next_rid;
  		t_key.flags = DB_DBT_USERMEM;
  		t_data.size = left;
  		if ((ret = __heapc_split(dbc, &t_key, &t_data, 0)) != 0)
  			goto err;
  
  		ACQUIRE_CUR(dbc,
  		    DB_LOCK_WRITE, last_rid.pgno, 0, DB_MPOOL_DIRTY, ret);
  		if (ret != 0)
  			goto err;
  
  		cp->indx = last_rid.indx;
  		old_hdr = (HEAPHDR *)(P_ENTRY(dbp, cp->page, cp->indx));
  
  		if (DBC_LOGGING(dbc)) {
  			old_size = DB_ALIGN(old_hdr->size +
  			    HEAP_HDRSIZE(old_hdr), sizeof(u_int32_t));
  			hdr_dbt.data = old_hdr;
  			hdr_dbt.size = HEAP_HDRSIZE(old_hdr);
  			log_dbt.data = (u_int8_t *)old_hdr + hdr_dbt.size;
  			log_dbt.size = DB_ALIGN(
  			    old_hdr->size, sizeof(u_int32_t));
  			if ((ret = __heap_addrem_log(dbp, dbc->txn,
  			    &LSN(cp->page), 0,
  			    OP_SET(DB_REM_HEAP, cp->page), cp->pgno,
  			    (u_int32_t)cp->indx, old_size,
  			    &hdr_dbt, &log_dbt, &LSN(cp->page))) != 0)
  				goto err;
  		} else
  			LSN_NOT_LOGGED(LSN(cp->page));
  
  		((HEAPSPLITHDR *)old_hdr)->nextpg = next_rid.pgno;
  		((HEAPSPLITHDR *)old_hdr)->nextindx = next_rid.indx;
  
  		if (DBC_LOGGING(dbc)) {
  			if ((ret = __heap_addrem_log(dbp, dbc->txn,
  			    &LSN(cp->page), 0,
  			    OP_SET(DB_ADD_HEAP, cp->page), cp->pgno,
  			    (u_int32_t)cp->indx, old_size,
  			    &hdr_dbt, &log_dbt, &LSN(cp->page))) != 0)
  				goto err;
  		} else
  			LSN_NOT_LOGGED(LSN(cp->page));
  
  		DISCARD(dbc, cp->page, cp->lock, 1, ret);
  	}
  
  err:	DB_ASSERT(dbp->env, ret != DB_PAGE_NOTFOUND);
  	if (buf != NULL)
  		__os_free(dbp->env, buf);
  	return (ret);
  }
  
  /*
   * __heapc_reloc --
   *	 Move data from a too-full page to a new page.  The old data page must
   *	 be write locked before calling this method.
   */
  static int
  __heapc_reloc(dbc, key, data)
  	DBC *dbc;
  	DBT *key;
  	DBT *data;
  {
  	DB *dbp;
  	DBT hdr_dbt, log_dbt, t_data, t_key;
  	DB_HEAP_RID last_rid, next_rid;
  	HEAPHDR *old_hdr;
  	HEAPSPLITHDR new_hdr;
  	HEAP_CURSOR *cp;
  	int is_first, ret;
  	u_int32_t left, old_size, size;
  
  	dbp = dbc->dbp;
  	cp = (HEAP_CURSOR *)dbc->internal;
  	old_hdr = (HEAPHDR *)(P_ENTRY(dbp, cp->page, cp->indx));
  	memset(&hdr_dbt, 0, sizeof(DBT));
  	memset(&log_dbt, 0, sizeof(DBT));
  	COMPQUIET(key, NULL);
  	/* Blob database records never change size. */
  	DB_ASSERT(dbp->env, !F_ISSET(old_hdr, HEAP_RECBLOB));
  
  	/*
  	 * We are updating an existing record, which will grow into a split
  	 * record.  The strategy is to overwrite the existing record (or each
  	 * piece of the record if the record is already split.)  If the new
  	 * record is shorter than the old, delete any extra pieces.  If the new
  	 * record is longer than the old, use heapc_split() to write the extra
  	 * data.
  	 *
  	 * We start each loop with t_data.data positioned to the next byte to be
  	 * written, old_hdr pointed at the header for the old record and the
  	 * necessary page write locked in cp->page.
  	 */
  	is_first = 1;
  	left = data->size;
  	memset(&t_data, 0, sizeof(DBT));
  	t_data.data = data->data;
  	for (;;) {
  		/* Figure out if we have a next piece. */
  		if (F_ISSET(old_hdr, HEAP_RECSPLIT)) {
  			next_rid.pgno = ((HEAPSPLITHDR *)old_hdr)->nextpg;
  			next_rid.indx = ((HEAPSPLITHDR *)old_hdr)->nextindx;
  		} else {
  			next_rid.pgno = PGNO_INVALID;
  			next_rid.indx = 0;
  		}
  
  		/* Delete the old data, after logging it. */
  		old_size = DB_ALIGN(
  		    old_hdr->size + HEAP_HDRSIZE(old_hdr), sizeof(u_int32_t));
  		if (old_size < sizeof(HEAPSPLITHDR))
  			old_size = sizeof(HEAPSPLITHDR);
  		if (DBC_LOGGING(dbc)) {
  			hdr_dbt.data = old_hdr;
  			hdr_dbt.size = HEAP_HDRSIZE(old_hdr);
  			log_dbt.data = (u_int8_t *)old_hdr + hdr_dbt.size;
  			log_dbt.size = DB_ALIGN(
  			    old_hdr->size, sizeof(u_int32_t));
  			if ((ret = __heap_addrem_log(dbp, dbc->txn,
  			    &LSN(cp->page), 0,
  			    OP_SET(DB_REM_HEAP, cp->page), cp->pgno,
  			    (u_int32_t)cp->indx, old_size,
  			    &hdr_dbt, &log_dbt, &LSN(cp->page))) != 0)
  				goto err;
  		} else
  			LSN_NOT_LOGGED(LSN(cp->page));
  		if ((ret = __heap_ditem(
  		    dbc, cp->page, cp->indx, old_size)) != 0)
  			goto err;
  
  		if (left == 0)
  			/*
  			 * We've finished writing the new record, we're just
  			 * cleaning up the old record now.
  			 */
  			goto next_pg;
  
  		/* Set up the header for the new record. */
  		memset(&new_hdr, 0, sizeof(HEAPSPLITHDR));
  		new_hdr.std_hdr.flags = HEAP_RECSPLIT;
  		/* We'll set this later if next_rid.pgno == PGNO_INVALID. */
  		new_hdr.nextpg = next_rid.pgno;
  		new_hdr.nextindx = next_rid.indx;
  		/*
  		 * Figure out how much we can fit on the page, rounding down to
  		 * a multiple of 4.  If we will have to expand the offset table,
  		 * account for that.It needs to be enough to at least fit the
  		 * split header.
  		 */
  		size = HEAP_FREESPACE(dbp, cp->page);
  		if (NUM_ENT(cp->page) == 0 ||
  		    cp->indx > HEAP_HIGHINDX(cp->page))
  			size -= sizeof(db_indx_t);
  		/* Round down to a multiple of 4. */
  		size = DB_ALIGN(
  		    (size - sizeof(u_int32_t)) + 1, sizeof(u_int32_t));
  		DB_ASSERT(dbp->env, size >= sizeof(HEAPSPLITHDR));
  		new_hdr.std_hdr.size =
  		    (u_int16_t)(size - sizeof(HEAPSPLITHDR));
  		if (new_hdr.std_hdr.size >= left) {
  			new_hdr.std_hdr.size = left;
  			new_hdr.std_hdr.flags |= HEAP_RECLAST;
  			new_hdr.nextpg = PGNO_INVALID;
  			new_hdr.nextindx = 0;
  		}
  		if (is_first) {
  			new_hdr.std_hdr.flags |= HEAP_RECFIRST;
  			new_hdr.tsize = left;
  			is_first = 0;
  		}
  
  		/* Now write the new data to the page. */
  		t_data.size = new_hdr.std_hdr.size;
  		hdr_dbt.data = &new_hdr;
  		hdr_dbt.size = sizeof(HEAPSPLITHDR);
  		/* Log the write. */
  		if (DBC_LOGGING(dbc)) {
  			if ((ret = __heap_addrem_log(dbp,
  			    dbc->txn, &LSN(cp->page), 0,
  			    OP_SET(DB_ADD_HEAP, cp->page),
  			    cp->pgno, (u_int32_t)cp->indx,
  			    size, &hdr_dbt, &t_data, &LSN(cp->page))) != 0)
  				goto err;
  		} else
  			LSN_NOT_LOGGED(LSN(cp->page));
  		if ((ret = __heap_pitem(dbc,
  		    (PAGE *)cp->page, cp->indx, size, &hdr_dbt, &t_data)) != 0)
  			goto err;
  
  		left -= new_hdr.std_hdr.size;
  		t_data.data = (u_int8_t *)(t_data.data) + new_hdr.std_hdr.size;
  
  		/* Get the next page, if any. */
  next_pg:	if (next_rid.pgno != PGNO_INVALID) {
  			ACQUIRE_CUR(dbc, DB_LOCK_WRITE,
  			    next_rid.pgno, 0, DB_MPOOL_DIRTY, ret);
  			if (ret != 0)
  				goto err;
  			cp->indx = next_rid.indx;
  			old_hdr = (HEAPHDR *)(P_ENTRY(dbp, cp->page, cp->indx));
  		} else {
  			/*
  			 * Remember the final piece's RID, we may need to update
  			 * the header after writing the rest of the record.
  			 */
  			last_rid.pgno = cp->pgno;
  			last_rid.indx = cp->indx;
  			/* Discard the page and drop the lock, txn-ally. */
  			DISCARD(dbc, cp->page, cp->lock, 1, ret);
  			if (ret != 0)
  				goto err;
  			break;
  		}
  	}
  
  	/*
  	 * If there is more work to do, let heapc_split do it.  After
  	 * heapc_split returns we need to update nextpg and nextindx in the
  	 * header of the last piece we wrote above.
  	 *
  	 * For logging purposes, we "delete" the old record and then "add" the
  	 * record.  This makes redo/undo work as-is, but we won't actually
  	 * delete and re-add the record.
  	 */
  	if (left > 0) {
  		memset(&t_key, 0, sizeof(DBT));
  		t_key.size = t_key.ulen = sizeof(DB_HEAP_RID);
  		t_key.data = &next_rid;
  		t_key.flags = DB_DBT_USERMEM;
  		t_data.size = left;
  		if ((ret = __heapc_split(dbc, &t_key, &t_data, 0)) != 0)
  			goto err;
  
  		ACQUIRE_CUR(dbc,
  		    DB_LOCK_WRITE, last_rid.pgno, 0, DB_MPOOL_DIRTY, ret);
  		if (ret != 0)
  			goto err;
  
  		cp->indx = last_rid.indx;
  		old_hdr = (HEAPHDR *)(P_ENTRY(dbp, cp->page, cp->indx));
  
  		if (DBC_LOGGING(dbc)) {
  			old_size = DB_ALIGN(old_hdr->size +
  			    HEAP_HDRSIZE(old_hdr), sizeof(u_int32_t));
  			hdr_dbt.data = old_hdr;
  			hdr_dbt.size = HEAP_HDRSIZE(old_hdr);
  			log_dbt.data = (u_int8_t *)old_hdr + hdr_dbt.size;
  			log_dbt.size = DB_ALIGN(
  			    old_hdr->size, sizeof(u_int32_t));
  			if ((ret = __heap_addrem_log(dbp, dbc->txn,
  			    &LSN(cp->page), 0,
  			    OP_SET(DB_REM_HEAP, cp->page), cp->pgno,
  			    (u_int32_t)cp->indx, old_size,
  			    &hdr_dbt, &log_dbt, &LSN(cp->page))) != 0)
  				goto err;
  		} else
  			LSN_NOT_LOGGED(LSN(cp->page));
  
  		((HEAPSPLITHDR *)old_hdr)->nextpg = next_rid.pgno;
  		((HEAPSPLITHDR *)old_hdr)->nextindx = next_rid.indx;
  
  		if (DBC_LOGGING(dbc)) {
  			if ((ret = __heap_addrem_log(dbp, dbc->txn,
  			    &LSN(cp->page), 0,
  			    OP_SET(DB_ADD_HEAP, cp->page), cp->pgno,
  			    (u_int32_t)cp->indx,old_size,
  			    &hdr_dbt, &log_dbt, &LSN(cp->page))) != 0)
  				goto err;
  		} else
  			LSN_NOT_LOGGED(LSN(cp->page));
  
  		DISCARD(dbc, cp->page, cp->lock, 1, ret);
  	}
  
  err:	DB_ASSERT(dbp->env, ret != DB_PAGE_NOTFOUND);
  	return (ret);
  }
  
  /*
   * __heapc_put --
   *
   * Put using a cursor.  If the given key exists, update the associated data.  If
   * the given key does not exsist, return an error.
   */
  static int
  __heapc_put(dbc, key, data, flags, pgnop)
  	DBC *dbc;
  	DBT *key;
  	DBT *data;
  	u_int32_t flags;
  	db_pgno_t *pgnop;
  {
  	DB *dbp;
  	DBT hdr_dbt, log_dbt, new_data;
  	DB_MPOOLFILE *mpf;
  	HEAPBLOBHDR bhdr;
  	HEAPHDR hdr, *old_hdr;
  	HEAP_CURSOR *cp;
  	PAGE *rpage;
  	db_pgno_t region_pgno;
  	int buf_alloc, ret, t_ret;
  	off_t blob_size;
  	db_seq_t blob_id, new_blob_id;
  	u_int32_t data_size, dlen, new_size, old_flags, old_size;
  	u_int32_t oldspace, space, tot_size;
  	u_int8_t *buf, *olddata;
  
  	dbp = dbc->dbp;
  	mpf = dbp->mpf;
  	cp = (HEAP_CURSOR *)dbc->internal;
  	rpage = NULL;
  	buf = NULL;
  	buf_alloc = 0;
  	dlen = 0;
  	blob_id = new_blob_id = 0;
  
  	if (flags != DB_CURRENT) {
  		/* We're going to write following the get, so use RMW. */
  		old_flags = dbc->flags;
  		F_SET(dbc, DBC_RMW);
  		ret = __heapc_get(dbc, key, data, DB_SET, pgnop);
  		F_CLR(key, DB_DBT_ISSET);
  		dbc->flags = old_flags;
  		DB_ASSERT(dbp->env, ret != DB_PAGE_NOTFOUND);
  		if (ret != 0)
  			return (ret);
  		else if (flags == DB_NOOVERWRITE)
  			return (DB_KEYEXIST);
  		if ((ret = __memp_dirty(mpf, &cp->page,
  		    dbc->thread_info, dbc->txn, dbc->priority, 0)) != 0)
  			return (ret);
  	} else {
  		/* We have a read lock, but need a write lock. */
  		if (STD_LOCKING(dbc) && cp->lock_mode != DB_LOCK_WRITE &&
  		    (ret = __db_lget(dbc,
  		    LCK_COUPLE, cp->pgno, DB_LOCK_WRITE, 0, &cp->lock)) != 0)
  			return (ret);
  
  		if ((ret = __memp_fget(mpf, &cp->pgno, dbc->thread_info,
  		    dbc->txn, DB_MPOOL_DIRTY, &cp->page)) != 0)
  			return (ret);
  	}
  
  	/* We've got the page locked and stored in cp->page. */
  	HEAP_CALCSPACEBITS(dbp, HEAP_FREESPACE(dbp, cp->page), oldspace);
  
  	/*
  	 * Figure out the spacing issue.  There is a very rare corner case where
  	 * we don't have enough space on the page to expand the data. Splitting
  	 * the record results in a larger header, if the page is jam packed
  	 * there might not be room for the larger header.
  	 *
  	 * hdr->size is the size of the stored data, it doesn't include any
  	 * padding.
  	 */
  	old_hdr = (HEAPHDR *)(P_ENTRY(dbp, cp->page, cp->indx));
  	/* Need data.size + header size, 4-byte aligned. */
  	old_size =
  	    DB_ALIGN(old_hdr->size + HEAP_HDRSIZE(old_hdr), sizeof(u_int32_t));
  	if (old_size < sizeof(HEAPSPLITHDR))
  		old_size = sizeof(HEAPSPLITHDR);
  	/* Partial puts on blobs are dealt with in the blob code. */
  	if (F_ISSET(data, DB_DBT_PARTIAL) && !F_ISSET(old_hdr, HEAP_RECBLOB)) {
  		if (F_ISSET(old_hdr, HEAP_RECSPLIT))
  			tot_size = ((HEAPSPLITHDR *)old_hdr)->tsize;
  		else
  			tot_size = old_hdr->size;
  		if (tot_size < data->doff) {
  			/* Post-pending */
  			dlen = data->dlen;
  			data_size = data->doff + data->size;
  		} else {
  			if (tot_size - data->doff < data->dlen)
  				dlen = tot_size - data->doff;
  			else
  				dlen = data->dlen;
  			data_size = (tot_size - dlen) + data->size;
  		}
  	} else if F_ISSET(old_hdr, HEAP_RECBLOB)
  		data_size = HEAPBLOBREC_DSIZE;
  	else
  		data_size = data->size;
  	new_size = DB_ALIGN(data_size + sizeof(HEAPHDR), sizeof(u_int32_t));
  	if (new_size < sizeof(HEAPSPLITHDR))
  		new_size = sizeof(HEAPSPLITHDR);
  
  	/* Check whether we actually have enough space on this page. */
  	if (F_ISSET(old_hdr, HEAP_RECSPLIT) ||
  	    (new_size > old_size &&
  	    new_size - old_size > HEAP_FREESPACE(dbp, cp->page))) {
  		/* Blob database records never change size. */
  		DB_ASSERT(dbp->env, !F_ISSET(old_hdr, HEAP_RECBLOB));
  		/*
  		 * We've got to split the record, not enough room on the
  		 * page.  Splitting the record will remove old_size bytes and
  		 * introduce at least sizeof(HEAPSPLITHDR).
  		 */
  		if (F_ISSET(data, DB_DBT_PARTIAL))
  			return (__heapc_reloc_partial(dbc, key, data));
  		else
  			return (__heapc_reloc(dbc, key, data));
  	}
  
  	memset(&new_data, 0, sizeof(DBT));
  	new_data.size = data_size;
  	if (F_ISSET(data, DB_DBT_PARTIAL) && !F_ISSET(old_hdr, HEAP_RECBLOB)) {
  		/*
  		 * Before replacing the old data, we need to use it to build the
  		 * new data.
  		 */
  		if ((ret = __os_malloc(dbp->env, data_size, &buf)) != 0)
  			goto err;
  		buf_alloc = 1;
  		new_data.data = buf;
  
  		/*
  		 * Preserve data->doff bytes at the start, or all of the old
  		 * record plus padding, if post-pending.
  		 */
  		olddata = (u_int8_t *)old_hdr + sizeof(HEAPHDR);
  		if (data->doff > old_hdr->size) {
  			memcpy(buf, olddata, old_hdr->size);
  			buf += old_hdr->size;
  			memset(buf, '\0', data->doff - old_hdr->size);
  			buf += data->doff - old_hdr->size;
  		} else {
  			memcpy(buf, olddata, data->doff);
  			buf += data->doff;
  		}
  
  		/* Now copy in the user's data. */
  		memcpy(buf, data->data, data->size);
  		buf += data->size;
  
  		/* Fill in remaining data from the old record, skipping dlen. */
  		if ((data->doff + data->dlen) < old_hdr->size) {
  			olddata += data->doff + data->dlen;
  			memcpy(buf, olddata,
  			    (old_hdr->size - data->doff) - data->dlen);
  		}
  	} else if (F_ISSET(old_hdr, HEAP_RECBLOB)) {
  		data_size = HEAPBLOBREC_DSIZE;
  		new_data.size = HEAPBLOBREC_DSIZE;
  		if (F_ISSET(data, DB_DBT_BLOB_REC)) {
  			DB_ASSERT(dbp->env,
  			    F_ISSET(((HEAPHDR *)data->data), HEAP_RECBLOB));
  			new_data.data = HEAPBLOBREC_DATA(data->data);
  		} else {
  			memcpy(&bhdr, old_hdr, HEAPBLOBREC_SIZE);
  			blob_id = (db_seq_t)bhdr.id;
  			GET_BLOB_SIZE(dbp->env, bhdr, blob_size, ret);
  			if (ret != 0)
  				goto err;
  			if ((ret = __blob_repl(dbc,
  			    data, blob_id, &new_blob_id, &blob_size)) != 0)
  				goto err;
  			bhdr.std_hdr.flags = HEAP_RECBLOB;
  			bhdr.std_hdr.size = HEAPBLOBREC_DSIZE;
  			SET_BLOB_SIZE(&bhdr, blob_size, HEAPBLOBHDR);
  			SET_BLOB_ID(&bhdr, new_blob_id, HEAPBLOBHDR);
  			new_data.data = HEAPBLOBREC_DATA(&bhdr);
  		}
  	} else {
  		new_data.data = data->data;
  	}
  
  	/*
  	 * Do the update by deleting the old record and writing the new
  	 * record.  Start by logging the entire operation.
  	 */
  	memset(&hdr, 0, sizeof(HEAPHDR));
  	hdr.size = data_size;
  	if (F_ISSET(old_hdr, HEAP_RECBLOB))
  		hdr.flags = HEAP_RECBLOB;
  	if (DBC_LOGGING(dbc)) {
  		hdr_dbt.data = old_hdr;
  		hdr_dbt.size = HEAP_HDRSIZE(old_hdr);
  		log_dbt.data = (u_int8_t *)old_hdr + hdr_dbt.size;
  		log_dbt.size = DB_ALIGN(old_hdr->size, sizeof(u_int32_t));
  		if ((ret = __heap_addrem_log(dbp, dbc->txn, &LSN(cp->page),
  		    0, OP_SET(DB_REM_HEAP, cp->page), cp->pgno,
  		    (u_int32_t)cp->indx,
  		    old_size, &hdr_dbt, &log_dbt, &LSN(cp->page))) != 0)
  			goto err;
  		hdr_dbt.data = &hdr;
  		hdr_dbt.size = HEAP_HDRSIZE(&hdr);
  		if ((ret = __heap_addrem_log(dbp, dbc->txn, &LSN(cp->page),
  		    0, OP_SET(DB_ADD_HEAP, cp->page), cp->pgno,
  		    (u_int32_t)cp->indx,
  		    new_size, &hdr_dbt, &new_data, &LSN(cp->page))) != 0)
  			goto err;
  	} else
  		LSN_NOT_LOGGED(LSN(cp->page));
  
  	if ((ret = __heap_ditem(dbc, cp->page, cp->indx, old_size)) != 0)
  		goto err;
  	hdr_dbt.data = &hdr;
  	hdr_dbt.size = HEAP_HDRSIZE(&hdr);
  	if ((ret = __heap_pitem(dbc,
  	    (PAGE *)cp->page, cp->indx, new_size, &hdr_dbt, &new_data)) != 0)
  		goto err;
  
  	/* Check whether we need to update the space bitmap. */
  	HEAP_CALCSPACEBITS(dbp, HEAP_FREESPACE(dbp, cp->page), space);
  
  	if (space != oldspace) {
  		/* Get the region page with an exclusive latch. */
  		region_pgno = HEAP_REGION_PGNO(dbp, cp->pgno);
  
  		if ((ret = __memp_fget(mpf, &region_pgno,
  		    dbc->thread_info, NULL, DB_MPOOL_DIRTY, &rpage)) != 0)
  			goto err;
  
  		HEAP_SETSPACE(dbp, rpage, (cp->pgno - region_pgno) - 1, space);
  	}
  
  err:	DB_ASSERT(dbp->env, ret != DB_PAGE_NOTFOUND);
  	if (rpage != NULL && (t_ret = __memp_fput(mpf,
  	    dbc->thread_info, rpage, dbc->priority)) != 0 && ret == 0)
  		ret = t_ret;
  	if (buf_alloc)
  		__os_free(dbp->env, new_data.data);
  
  	if (ret != 0 && LOCK_ISSET(cp->lock))
  		(void)__TLPUT(dbc, cp->lock);
  
  	return (ret);
  }
  
  /*
   * __heap_getpage --
   *	Return a page with sufficient free space.  The page will be write locked
   *	and marked dirty.
   */
  static int
  __heap_getpage(dbc, size, avail)
  	DBC *dbc;
  	u_int32_t size;
  	u_int8_t *avail;
  {
  	DB *dbp;
  	DBMETA *meta;
  	DB_LOCK meta_lock;
  	DB_LSN meta_lsn;
  	DB_MPOOLFILE *mpf;
  	HEAP *h;
  	HEAPPG *rpage;
  	HEAP_CURSOR *cp;
  	db_pgno_t data_pgno, i, max, meta_pgno, p, region_pgno, start;
  	db_pgno_t start_region;
  	int ret, t_ret;
  	u_int32_t lk_mode, space;
  
  	LOCK_INIT(meta_lock);
  	data_pgno = PGNO_INVALID;
  	dbp = dbc->dbp;
  	mpf = dbp->mpf;
  	cp = (HEAP_CURSOR *)dbc->internal;
  	h = dbp->heap_internal;
  	start_region = region_pgno = h->curregion;
  	max = HEAP_REGION_SIZE(dbp);
  	i = 0;
  	ret = t_ret = 0;
  
  	/*
  	 * The algorithm for finding a page:
  	 *
  	 * Look in the space bitmap of the current region page for a data page
  	 * with at least size bytes free.  Once we find a page, try to lock it
  	 * and if we get the lock we're done.
  	 *
  	 * Don't wait for a locked region page, just move on to the next region
  	 * page, creating it if it doesn't exist.  If the size of the heap
  	 * database is not constrained, just keep creating regions and extending
  	 * the database until we find a page with space.  If the database size
  	 * is constrained, loop back to the first region page from the final
  	 * region page.  If we wind up making it all the way back to where our
  	 * search began, we need to start waiting for locked region pages.  If
  	 * we finish another loop through the database waiting for every region
  	 * page, we know there's no room.
  	 */
  
  	/*
  	 * Figure out the % of the page the data will occupy and translate that
  	 * to the relevant bit-map value we need to look for.
  	 */
  	HEAP_CALCSPACEBITS(dbp, size, space);
  
  	/*
  	 * Get the current region page, with a shared latch.  On the first loop
  	 * through a fixed size database, we move on to the next region if the
  	 * page is locked.  On the second loop, we wait for locked region
  	 * pages.  If the database isn't fixed size, we never wait, we'll
  	 * eventually get to use one of the region pages we create.
  	 */
  	lk_mode = DB_MPOOL_TRY;
  find:	while ((ret = __memp_fget(mpf, &region_pgno,
  	    dbc->thread_info, NULL, lk_mode, &rpage)) != 0 ||
  	    TYPE(rpage) != P_IHEAP) {
  		if (ret == DB_LOCK_NOTGRANTED)
  			goto next_region;
  		if (ret != 0 && ret != DB_PAGE_NOTFOUND)
  			return (ret);
  		/*
  		 * The region page doesn't exist, or hasn't been initialized,
  		 * create it, then try again.  If the page exists, we have to
  		 * drop it before initializing the region.
  		 */
  		if (ret == 0 && (ret = __memp_fput(
  		    mpf, dbc->thread_info, rpage, dbc->priority)) != 0)
  			return (ret);
  
  		if ((ret = __heap_create_region(dbc, region_pgno)) != 0)
  			return (ret);
  	}
  
  	start = h->curpgindx;
  	/*
  	 * If this is the last region page in a fixed size db, figure out the
  	 * maximum pgno in the bitmap.
  	 */
  	if (region_pgno + max > h->maxpgno)
  		max = h->maxpgno - region_pgno;
  	/*
  	 * Look in the bitmap for a page with sufficient free space.  We use i
  	 * in a slightly strange way.  Because the 2-bits in the bitmap are
  	 * only an estimate, there is a chance the data won't fit on the page
  	 * we choose.  In that case, we re-start the process and want to be
  	 * able to resume this loop where we left off.
  	 */
  	for (; i < max; i++) {
  		p = start + i;
  		if (p >= max)
  			p -= max;
  		if ((*avail = HEAP_SPACE(dbp, rpage, p)) > space)
  			continue;
  		data_pgno = (region_pgno + p) + 1;
  		ACQUIRE_CUR(dbc,
  		    DB_LOCK_WRITE, data_pgno, DB_LOCK_NOWAIT, 0, ret);
  		/*
  		 * If we have the lock and the page or have the lock and need to
  		 * create the page, we're good.  If we don't have the lock, try
  		 * to find different page.
  		 */
  		if (ret == 0 || ret == DB_PAGE_NOTFOUND)
  			break;
  		else if (ret == DB_LOCK_NOTGRANTED || ret == DB_LOCK_DEADLOCK) {
  			ret = 0;
  			continue;
  		} else
  			goto err;
  	}
  
  	/*
  	 * Keep a worst case range of highest used page in the region.
  	 */
  	if (i < max && data_pgno > rpage->high_pgno) {
  		if ((ret = __memp_dirty(mpf,
  		    &rpage, dbc->thread_info, NULL, dbc->priority, 0)) != 0)
  			goto err;
  		/* We might have blocked, check again */
  		if (data_pgno > rpage->high_pgno)
  			rpage->high_pgno = data_pgno;
  	}
  
  	/* Done with the region page, even if we didn't find a page. */
  	if ((ret = __memp_fput(mpf,
  	    dbc->thread_info, rpage, dbc->priority)) != 0) {
  		/* Did not read the data page, so we can release its lock. */
  		DISCARD(dbc, cp->page, cp->lock, 0, t_ret);
  		goto err;
  	}
  	rpage = NULL;
  
  	if (i >= max) {
  		/*
  		 * No free pages on this region page, advance to the next region
  		 * page.  If we're at the end of a fixed size heap db, loop
  		 * around to the first region page.  There is not currently a
  		 * data page locked.
  		 */
  next_region:	region_pgno += HEAP_REGION_SIZE(dbp) + 1;
  
  		if (region_pgno > h->maxpgno)
  			region_pgno = FIRST_HEAP_RPAGE;
  
  		if (region_pgno == start_region) {
  			/*
  			 * We're in a fixed size db and we've looped through all
  			 * region pages.
  			 */
  
  			if (lk_mode == DB_MPOOL_TRY) {
  				/*
  				 * We may have missed a region page with room,
  				 * because we didn't wait for locked pages.  Try
  				 * another loop, waiting for all pages.
  				 */
  				lk_mode = 0;
  			} else {
  				/*
  				 * We've seen every region page, because we
  				 * waited for all pages.  No room.
  				 */
  				ret = DB_HEAP_FULL;
  				goto err;
  			}
  		}
  
  		h->curregion = region_pgno;
  		h->curpgindx = 0;
  		i = 0;
  		goto find;
  	}
  
  	/*
  	 * At this point we have the page locked.  If we have the page, we need
  	 * to mark it dirty.  If we don't have the page (or if the page is
  	 * empty) we need to create and initialize it.
  	 */
  	if (cp->pgno == PGNO_INVALID || PGNO(cp->page) == PGNO_INVALID) {
  		/*
  		 * The data page needs to be created and the metadata page needs
  		 * to be updated.  Once we get the metadata page, we must not
  		 * jump to err, the metadata page and lock are put back here.
  		 *
  		 * It is possible that the page was created by an aborted txn,
  		 * in which case the page exists but is all zeros.  We still
  		 * need to "create" it and log the creation.
  		 *
  		 */
  
  		meta_pgno = PGNO_BASE_MD;
  		if ((ret = __db_lget(dbc, LCK_ALWAYS, meta_pgno,
  		    DB_LOCK_WRITE, DB_LOCK_NOWAIT, &meta_lock)) != 0) {
  			/*
  			 * We don't want to block while having latched
  			 * a page off the end of file.  This could
  			 * get truncated by another thread and we
  			 * will deadlock.
  			 */
  			p = cp->page != NULL;
  			DISCARD(dbc, cp->page, cp->lock, 0, t_ret);
  			if (t_ret != 0 ||
  			     (ret != DB_LOCK_NOTGRANTED &&
  			     ret != DB_LOCK_DEADLOCK))
  				goto pg_err;
  			if ((ret = __db_lget(dbc, LCK_ALWAYS, meta_pgno,
  			    DB_LOCK_WRITE, 0, &meta_lock)) != 0)
  				goto pg_err;
  			ACQUIRE_CUR(dbc, DB_LOCK_WRITE,
  			    data_pgno, 0, DB_MPOOL_CREATE, ret);
  			/*
  			 * We can race, having read this page when it was
  			 * less than last_pgno but now an aborted
  			 * allocation can make this page beyond last_pgno
  			 * so we must free it. If we can't get the
  			 * lock on the page again, then some other
  			 * thread will handle the issue.
  			 */
  			if (ret != 0) {
  pg_err:				if (p != 0) {
  					ACQUIRE_CUR(dbc, DB_LOCK_WRITE,
  					    data_pgno, 0, 0, t_ret);
  					if (t_ret == 0 &&
  					     PGNO(cp->page) == PGNO_INVALID) {
  						(void)__memp_fput(mpf,
  						     dbc->thread_info,
  						     cp->page, dbc->priority);
  						(void)__memp_fget(mpf,
  						     &data_pgno,
  						     dbc->thread_info, dbc->txn,
  						     DB_MPOOL_FREE, &cp->page);
  					}
  					(void)__LPUT(dbc, cp->lock);
  				}
  				(void)__LPUT(dbc, meta_lock);
  				goto err;
  			}
  			/* Check if we lost a race. */
  			if (PGNO(cp->page) != PGNO_INVALID) {
  				if ((ret = __LPUT(dbc, meta_lock)) != 0)
  					goto err;
  				goto check;
  			}
  		}
  
  		/*
  		 * Before creating a new page in this region, check that the
  		 * region page still exists.  By this point, the transaction
  		 * that created the region must have aborted or committed,
  		 * because we now hold the metadata lock.  If we can't get the
  		 * latch, the page must exist.
  		 */
  		ret = __memp_fget(mpf, &region_pgno,
  		    dbc->thread_info, NULL, DB_MPOOL_TRY, &rpage);
  		if (ret == DB_LOCK_NOTGRANTED)
  			ret = 0;
  		else if (ret != 0) {
  			/*
  			 * Free up the metadata lock.  If this was an error
  			 * other than a missing region page, bail.
  			 */
  			if ((t_ret = __LPUT(dbc, meta_lock)) != 0)
  				ret = t_ret;
  			if (ret != DB_PAGE_NOTFOUND)
  				goto err;
  			/*
  			 * The region no longer exists.  Release the page's lock
  			 * (we haven't created the page yet) and find a new page
  			 * on a different region.
  			 */
  			DISCARD(dbc, cp->page, cp->lock, 0, t_ret);
  			goto find;
  		} else
  			ret = __memp_fput(mpf,
  			    dbc->thread_info, rpage, dbc->priority);
  		rpage = NULL;
  		if (ret != 0)
  			goto meta_unlock;
  
  		if ((ret = __memp_fget(mpf, &meta_pgno,
  		    dbc->thread_info, dbc->txn, DB_MPOOL_DIRTY, &meta)) != 0)
  			goto err;
  
  		/* Log the page creation.  Can't jump to err if it fails. */
  		if (DBC_LOGGING(dbc))
  			ret = __heap_pg_alloc_log(dbp,
  			    dbc->txn, &LSN(meta), 0, &LSN(meta), meta_pgno,
  				data_pgno, (u_int32_t)P_HEAP, meta->last_pgno);
  		else
  			LSN_NOT_LOGGED(LSN(meta));
  
  		/*
  		 * We may have created a page earlier with a larger page number
  		 * check before updating the metadata page.
  		 */
  		if (ret == 0 && data_pgno > meta->last_pgno)
  			meta->last_pgno = data_pgno;
  		meta_lsn = LSN(meta);
  
  		if ((t_ret = __memp_fput(mpf,
  		    dbc->thread_info, meta, dbc->priority)) != 0 && ret == 0)
  			ret = t_ret;
  		meta = NULL;
  		if (ret != 0)
  			goto meta_unlock;
  
  		/* If the page doesn't actually exist we need to create it. */
  		if (cp->pgno == PGNO_INVALID) {
  			cp->pgno = data_pgno;
  			if ((ret = __memp_fget(mpf, &cp->pgno,
  			    dbc->thread_info, dbc->txn,
  			    DB_MPOOL_CREATE | DB_MPOOL_DIRTY, &cp->page)) != 0)
  				goto meta_unlock;
  			DB_ASSERT(dbp->env, cp->pgno == data_pgno);
  		} else if ((ret = __memp_dirty(mpf, &cp->page,
  		    dbc->thread_info, dbc->txn, dbc->priority, 0)) != 0) {
  			/* Did not read the page, so we can release the lock. */
  			DISCARD(dbc, cp->page, cp->lock, 0, t_ret);
  			goto meta_unlock;
  		}
  
  		/* Now that we have the page we initialize it and we're done. */
  		P_INIT(cp->page,
  		    dbp->pgsize, cp->pgno, P_INVALID, P_INVALID, 0, P_HEAP);
  		LSN(cp->page) = meta_lsn;
  
  meta_unlock:	if ((t_ret = __TLPUT(dbc, meta_lock)) != 0 && ret == 0)
  			ret = t_ret;
  		if (ret != 0)
  			goto err;
  	} else {
  		/* Check whether we actually have enough space on this page. */
  check:		if (size + sizeof(db_indx_t) > HEAP_FREESPACE(dbp, cp->page)) {
  			/* Put back the page and lock, they were never used. */
  			DISCARD(dbc, cp->page, cp->lock, 0, ret);
  			if (ret != 0)
  				goto err;
  
  			/* Re-start the bitmap check on the next page. */
  			i++;
  			goto find;
  		}
  
  		if ((ret = __memp_dirty(mpf, &cp->page,
  		    dbc->thread_info, dbc->txn, dbc->priority, 0)) != 0) {
  			/* Did not read the page, so we can release the lock. */
  			DISCARD(dbc, cp->page, cp->lock, 0, t_ret);
  			goto err;
  		}
  	}
  
  	h->curpgindx = (data_pgno - region_pgno) - 1;
  err:	DB_ASSERT(dbp->env, ret != DB_PAGE_NOTFOUND);
  	if (rpage != NULL && (t_ret = __memp_fput(mpf,
  	    dbc->thread_info, rpage, dbc->priority)) != 0 && ret == 0)
  		ret = t_ret;
  
  	return (ret);
  }
  
  /*
   * __heap_append --
   *	Add an item to a heap database.
   *
   * PUBLIC: int __heap_append
   * PUBLIC:     __P((DBC *, DBT *, DBT *));
   */
  int
  __heap_append(dbc, key, data)
  	DBC *dbc;
  	DBT *data, *key;
  {
  	DB *dbp;
  	DBT tmp_dbt, data_dbt;
  	DB_HEAP_RID rid;
  	DB_LSN lsn;
  	DB_MPOOLFILE *mpf;
  	HEAPPG *rpage;
  	HEAPBLOBHDR bhdr;
  	HEAPHDR hdr;
  	HEAP_CURSOR *cp;
  	db_indx_t indx;
  	db_pgno_t region_pgno;
  	int is_blob, ret, t_ret;
  	off_t blob_size;
  	db_seq_t blob_id;
  	u_int8_t avail;
  	u_int32_t data_size, space;
  
  	dbp = dbc->dbp;
  	mpf = dbp->mpf;
  	ret = t_ret = 0;
  	rpage = NULL;
  	cp = (HEAP_CURSOR *)dbc->internal;
  	blob_size = 0;
  	blob_id = 0;
  
  	if (dbp->blob_threshold &&
  	    (data->size >= dbp->blob_threshold || F_ISSET(data, DB_DBT_BLOB)))
  		is_blob = 1;
  	else
  		is_blob = 0;
  
  	/* Need data.size + header size, 4-byte aligned. */
  	if (is_blob)
  		data_size = HEAPBLOBREC_SIZE;
  	else if (F_ISSET(data, DB_DBT_PARTIAL))
  		data_size = DB_ALIGN(data->doff +
  		    data->size + sizeof(HEAPHDR), sizeof(u_int32_t));
  	else
  		data_size = DB_ALIGN(
  		    data->size + sizeof(HEAPHDR), sizeof(u_int32_t));
  
  	if (data_size >= HEAP_MAXDATASIZE(dbp))
  		return (__heapc_split(dbc, key, data, 1));
  	else if (data_size < sizeof(HEAPSPLITHDR))
  		data_size = sizeof(HEAPSPLITHDR);
  
  	if ((ret = __heap_getpage(dbc, data_size, &avail)) != 0)
  		goto err;
  
  	indx = HEAP_FREEINDX(cp->page);
  	if (is_blob) {
  		if ((ret = __blob_put(
  		    dbc, data, &blob_id, &blob_size, &lsn)) != 0)
  			goto err;
  		memset(&bhdr, 0, HEAPBLOBREC_SIZE);
  		bhdr.std_hdr.flags = HEAP_RECBLOB;
  		bhdr.std_hdr.size = HEAPBLOBREC_DSIZE;
  		SET_BLOB_SIZE(&bhdr, blob_size, HEAPBLOBHDR);
  		SET_BLOB_ID(&bhdr, blob_id, HEAPBLOBHDR);
  		SET_BLOB_FILE_ID(&bhdr, dbp->blob_file_id, HEAPBLOBHDR);
  		tmp_dbt.data = &bhdr;
  		tmp_dbt.size = sizeof(HEAPHDR);
  		memset(&data_dbt, 0, sizeof(DBT));
  		data_dbt.data = HEAPBLOBREC_DATA((&bhdr));
  		data_dbt.size = HEAPBLOBREC_DSIZE;
  	} else {
  		memset(&hdr, 0, sizeof(HEAPHDR));
  		hdr.size = data->size;
  		if (F_ISSET(data, DB_DBT_PARTIAL))
  			hdr.size += data->doff;
  		tmp_dbt.data = &hdr;
  		tmp_dbt.size = sizeof(HEAPHDR);
  		memcpy(&data_dbt, data, sizeof(DBT));
  	}
  
  	/* Log the write. */
  	if (DBC_LOGGING(dbc)) {
  		if ((ret = __heap_addrem_log(dbp, dbc->txn, &LSN(cp->page),
  		    0, OP_SET(DB_ADD_HEAP, cp->page), cp->pgno, (u_int32_t)indx,
  		    data_size, &tmp_dbt, &data_dbt, &LSN(cp->page))) != 0)
  			goto err;
  	} else
  		LSN_NOT_LOGGED(LSN(cp->page));
  
  	if ((ret = __heap_pitem(
  	    dbc, (PAGE *)cp->page, indx, data_size, &tmp_dbt, &data_dbt)) != 0)
  		goto err;
  
  	rid.pgno = cp->pgno;
  	rid.indx = indx;
  	cp->indx = indx;
  
  	/* Check whether we need to update the space bitmap. */
  	HEAP_CALCSPACEBITS(dbp, HEAP_FREESPACE(dbp, cp->page), space);
  
  	if (space != avail) {
  		/* Get the region page with an exclusive latch. */
  		region_pgno = HEAP_REGION_PGNO(dbp, cp->pgno);
  		if ((ret = __memp_fget(mpf, &region_pgno,
  		    dbc->thread_info, NULL, DB_MPOOL_DIRTY, &rpage)) != 0)
  			goto err;
  
  		HEAP_SETSPACE(dbp, rpage, (cp->pgno - region_pgno) - 1, space);
  	}
  
  err:	DB_ASSERT(dbp->env, ret != DB_PAGE_NOTFOUND);
  	if (rpage != NULL && (t_ret = __memp_fput(mpf,
  	    dbc->thread_info, rpage, dbc->priority)) != 0 && ret == 0)
  		ret = t_ret;
  
  	if (cp->page != NULL) {
  		DISCARD(dbc, cp->page, cp->lock, 1, t_ret);
  		if (ret == 0)
  			ret = t_ret;
  	}
  
  	if (ret == 0 && key != NULL)
  		ret = __db_retcopy(dbp->env, key,
  		    &rid, DB_HEAP_RID_SZ, &dbc->rkey->data, &dbc->rkey->ulen);
  
  	return (ret);
  }
  
  static int
  __heapc_split(dbc, key, data, is_first)
  	DBC *dbc;
  	DBT *key, *data;
  	int is_first;
  {
  	DB *dbp;
  	DBT hdr_dbt, t_data;
  	DB_HEAP_RID rid;
  	DB_MPOOLFILE *mpf;
  	HEAPPG *rpage;
  	HEAPSPLITHDR hdrs;
  	HEAP_CURSOR *cp;
  	db_indx_t indx;
  	db_pgno_t region_pgno;
  	int ret, t_ret;
  	u_int32_t buflen, doff, left, size, spacebits;
  	u_int8_t availbits, *buf;
  
  	dbp = dbc->dbp;
  	mpf = dbp->mpf;
  	cp = (HEAP_CURSOR *)dbc->internal;
  	memset(&hdrs, 0, sizeof(HEAPSPLITHDR));
  	memset(&t_data, 0, sizeof(DBT));
  	hdrs.std_hdr.flags = HEAP_RECSPLIT | HEAP_RECLAST;
  
  	doff = data->doff;
  	rpage = NULL;
  	ret = t_ret = 0;
  	indx = 0;
  	buf = NULL;
  
  	/*
  	 * Write the record to multiple pages, in chunks starting from the end.
  	 * To reconstruct during a get we need the RID of the next chunk, so if
  	 * work our way from back to front during writing we always know the rid
  	 * of the "next" chunk, it's the chunk we just wrote.
  	 */
  	t_data.data = (u_int8_t *)data->data + data->size;
  	left = data->size;
  	if (F_ISSET(data, DB_DBT_PARTIAL)) {
  		left += data->doff;
  	}
  	hdrs.tsize = left;
  	buflen = 1;
  	if ((ret = __os_malloc(dbp->env, buflen, &buf)) != 0)
  		return (ret);
  	while (left > 0) {
  		size = DB_ALIGN(left + sizeof(HEAPSPLITHDR), sizeof(u_int32_t));
  		if (size < sizeof(HEAPSPLITHDR))
  			size = sizeof(HEAPSPLITHDR);
  
  		if (size > HEAP_MAXDATASIZE(dbp))
  			/*
  			 * Data won't fit on a single page, find one at least
  			 * 33% free.
  			 */
  			size = DB_ALIGN(dbp->pgsize / 3, sizeof(u_int32_t));
  		else
  			hdrs.std_hdr.flags |= HEAP_RECFIRST;
  
  		if ((ret = __heap_getpage(dbc, size, &availbits)) != 0) {
  			__os_free(dbp->env, buf);
  			return (ret);
  		}
  
  		/*
  		 * size is the total number of bytes being written to the page.
  		 * The header holds the size of the data being written.
  		 */
  		if (F_ISSET(&(hdrs.std_hdr), HEAP_RECFIRST)) {
  			hdrs.std_hdr.size = left;
  			/*
  			 * If we're called from heapc_reloc, we are only writing
  			 * a piece of the full record and shouldn't set
  			 * HEAP_RECFIRST.
  			 */
  			if (!is_first)
  				F_CLR(&(hdrs.std_hdr), HEAP_RECFIRST);
  		} else {
  			/*
  			 * Figure out how much room is on the page.  If we will
  			 * have to expand the offset table, account for that.
  			 */
  			size = HEAP_FREESPACE(dbp, cp->page);
  			if (NUM_ENT(cp->page) == 0 ||
  			    HEAP_FREEINDX(cp->page) > HEAP_HIGHINDX(cp->page))
  				size -= sizeof(db_indx_t);
  			/* Round down to a multiple of 4. */
  			size = DB_ALIGN(
  			    (size - sizeof(u_int32_t)) + 1, sizeof(u_int32_t));
  			DB_ASSERT(dbp->env, size >= sizeof(HEAPSPLITHDR));
  			hdrs.std_hdr.size =
  			    (u_int16_t)(size - sizeof(HEAPSPLITHDR));
  		}
  
  		/*
  		 * t_data.data points at the end of the data left to write.  Now
  		 * that we know how much we're going to write to this page, we
  		 * can adjust the pointer to point at the start of the data to
  		 * be written.
  		 *
  		 * If DB_DBT_PARTIAL is set, once data->data is exhausted, we
  		 * have to pad with data->doff bytes (or as much as can fit on
  		 * this page.)  left - doff gives the number of bytes to use
  		 * from data->data.  Once that can't fill t_data, we have to
  		 * start padding.
  		 */
  		t_data.data = (u_int8_t *)(t_data.data) - hdrs.std_hdr.size;
  		DB_ASSERT(dbp->env, (F_ISSET(data, DB_DBT_PARTIAL) ||
  		    t_data.data >= data->data));
  		t_data.size = hdrs.std_hdr.size;
  		if (F_ISSET(data, DB_DBT_PARTIAL) &&
  		    t_data.size > left - doff) {
  			if (buflen < t_data.size) {
  				if (__os_realloc(
  				    dbp->env, t_data.size, &buf) != 0)
  					return (ENOMEM);
  				buflen = t_data.size;
  			}
  			/*
  			 * We have to figure out how much data remains.  left
  			 * includes doff, so we need (left - doff) bytes from
  			 * data.  We also need the amount of padding that can
  			 * fit on the page.  That's the amount we can fit on the
  			 * page minus the bytes we're taking from data.
  			*/
  			t_data.data = buf;
  			memset(buf, 0, (t_data.size - left) + doff);
  			buf += (t_data.size - left) + doff;
  			memcpy(buf, data->data, left - doff);
  			doff -= (t_data.size - left) + doff;
  			buf = t_data.data;
  		}
  		hdr_dbt.data = &hdrs;
  		hdr_dbt.size = sizeof(HEAPSPLITHDR);
  		indx = HEAP_FREEINDX(cp->page);
  
  		/* Log the write. */
  		if (DBC_LOGGING(dbc)) {
  			if ((ret = __heap_addrem_log(dbp,
  			    dbc->txn, &LSN(cp->page), 0,
  			    OP_SET(DB_ADD_HEAP, cp->page),
  			    cp->pgno, (u_int32_t)indx,
  			    size, &hdr_dbt, &t_data, &LSN(cp->page))) != 0)
  				goto err;
  		} else
  			LSN_NOT_LOGGED(LSN(cp->page));
  
  		if ((ret = __heap_pitem(dbc,
  		    (PAGE *)cp->page, indx, size, &hdr_dbt, &t_data)) != 0)
  			goto err;
  		F_CLR(&(hdrs.std_hdr), HEAP_RECLAST);
  		left -= hdrs.std_hdr.size;
  
  		/*
  		 * Save the rid where we just wrote, this is the "next"
  		 * chunk.
  		 */
  		hdrs.nextpg = cp->pgno;
  		hdrs.nextindx = indx;
  
  		/* Check whether we need to update the space bitmap. */
  		HEAP_CALCSPACEBITS(dbp,
  		    HEAP_FREESPACE(dbp, cp->page), spacebits);
  
  		if (spacebits != availbits) {
  			/* Get the region page with an exclusive latch. */
  			region_pgno = HEAP_REGION_PGNO(dbp, cp->pgno);
  			if ((ret = __memp_fget(mpf, &region_pgno,
  			    dbc->thread_info,
  			    NULL, DB_MPOOL_DIRTY, &rpage)) != 0)
  				goto err;
  
  			HEAP_SETSPACE(dbp,
  			    rpage, (cp->pgno - region_pgno) - 1, spacebits);
  			ret = __memp_fput(mpf,
  			    dbc->thread_info, rpage, dbc->priority);
  			rpage = NULL;
  			if (ret != 0)
  				goto err;
  		}
  
  	}
  
  	rid.pgno = cp->pgno;
  	rid.indx = indx;
  	cp->indx = indx;
  
  err:	if (rpage != NULL && (t_ret = __memp_fput(mpf,
  	    dbc->thread_info, rpage, dbc->priority)) != 0 && ret == 0)
  		ret = t_ret;
  	if (cp->page != NULL) {
  		DISCARD(dbc, cp->page, cp->lock, 1, t_ret);
  		if (ret == 0)
  			ret = t_ret;
  	}
  	if (buf != NULL)
  		__os_free(dbp->env, buf);
  
  	if (ret == 0 && key != NULL)
  		ret = __db_retcopy(dbp->env, key,
  		    &rid, DB_HEAP_RID_SZ, &dbc->rkey->data, &dbc->rkey->ulen);
  	DB_ASSERT(dbp->env, ret != DB_PAGE_NOTFOUND);
  	return (ret);
  }
  
  /*
   * __heapc_pitem --
   *	Put an item on a heap page.  Copy all bytes from the header (if any)
   *	first and then copy from data.
   *
   * PUBLIC: int __heap_pitem __P((DBC *,
   * PUBLIC:	PAGE *, u_int32_t, u_int32_t, DBT *, DBT *));
   */
  int
  __heap_pitem(dbc, pagep, indx, nbytes, hdr, data)
  	DBC *dbc;
  	PAGE *pagep;
  	u_int32_t indx;
  	u_int32_t nbytes;
  	DBT *hdr, *data;
  {
  	DB *dbp;
  	u_int8_t *buf;
  
  	dbp = dbc->dbp;
  
  	DB_ASSERT(dbp->env, TYPE(pagep) == P_HEAP);
  	DB_ASSERT(dbp->env, IS_DIRTY(pagep));
  	DB_ASSERT(dbp->env, nbytes == DB_ALIGN(nbytes, sizeof(u_int32_t)));
  	DB_ASSERT(dbp->env, DB_ALIGN(((HEAPHDR *)hdr->data)->size,
  	    sizeof(u_int32_t)) >= data->size);
  	DB_ASSERT(dbp->env, nbytes >= hdr->size + data->size);
  
  	/*
  	 * We're writing data either as a result of DB->put or as a result of
  	 * undo-ing a delete.	If we're undo-ing a delete we just need to write
  	 * the bytes from hdr to the page.  Otherwise, we need to construct a
  	 * heap header, etc.
  	 */
  	HEAP_OFFSETTBL(dbp, pagep)[indx] = HOFFSET(pagep) - nbytes;
  	buf = P_ENTRY(dbp, pagep, indx);
  	DB_ASSERT(dbp->env, buf > (u_int8_t*)&HEAP_OFFSETTBL(dbp, pagep)[indx]);
  
  	if (hdr != NULL) {
  		memcpy(buf, hdr->data, hdr->size);
  		buf += hdr->size;
  	}
  	if (F_ISSET(data, DB_DBT_PARTIAL)) {
  		memset(buf, 0, data->doff);
  		buf += data->doff;
  	}
  	memcpy(buf, data->data, data->size);
  
  	/*
  	 * Update data page header.  If DEBUG/DIAGNOSTIC is set, the page might
  	 * be filled with 0xdb, so we can't just look for a 0 in the offset
  	 * table.  We used the first available index, so start there and scan
  	 * forward.  If the table is full, the first available index is the
  	 * highest index plus one.
  	 */
  	if (indx > HEAP_HIGHINDX(pagep)) {
  		if (NUM_ENT(pagep) == 0)
  			HEAP_FREEINDX(pagep) = 0;
  		else if (HEAP_FREEINDX(pagep) >= indx) {
  			if (indx > (u_int32_t)HEAP_HIGHINDX(pagep) + 1)
  				HEAP_FREEINDX(pagep) = HEAP_HIGHINDX(pagep) + 1;
  			else
  				HEAP_FREEINDX(pagep) = indx + 1;
  		}
  		while (++HEAP_HIGHINDX(pagep) < indx)
  			HEAP_OFFSETTBL(dbp,pagep)[HEAP_HIGHINDX(pagep)] = 0;
  	} else {
  		for (; indx <= HEAP_HIGHINDX(pagep); indx++)
  			if (HEAP_OFFSETTBL(dbp, pagep)[indx] == 0)
  				break;
  		HEAP_FREEINDX(pagep) = indx;
  	}
  	HOFFSET(pagep) -= nbytes;
  	NUM_ENT(pagep)++;
  
  	return (0);
  }
  
  /*
   * __heapc_dup --
   *      Duplicate a heap cursor, such that the new one holds appropriate
   *      locks for the position of the original.
   *
   * PUBLIC: int __heapc_dup __P((DBC *, DBC *));
   */
  int
  __heapc_dup(orig_dbc, new_dbc)
  	DBC *orig_dbc, *new_dbc;
  {
  	HEAP_CURSOR *orig, *new;
  
  	orig = (HEAP_CURSOR *)orig_dbc->internal;
  	new = (HEAP_CURSOR *)new_dbc->internal;
  	new->flags = orig->flags;
  	return (0);
  }
  
  /*
   * __heapc_gsplit --
   *      Get a heap split record.  The page pointed to by the cursor must
   *	be the first segment of this record.
   *
   * PUBLIC: int __heapc_gsplit __P((DBC *,
   * PUBLIC:     DBT *, void **, u_int32_t *));
   */
  int
  __heapc_gsplit(dbc, dbt, bpp, bpsz)
  	DBC *dbc;
  	DBT *dbt;
  	void **bpp;
  	u_int32_t *bpsz;
  {
  	DB *dbp;
  	DB_MPOOLFILE *mpf;
  	DB_HEAP_RID rid;
  	DB_LOCK data_lock;
  	HEAP_CURSOR *cp;
  	ENV *env;
  	HEAPPG *dpage;
  	HEAPSPLITHDR *hdr;
  	db_indx_t bytes;
  	u_int32_t curoff, needed, start, tlen;
  	u_int8_t *p, *src;
  	int putpage, ret, t_ret;
  
  	LOCK_INIT(data_lock);
  	dbp = dbc->dbp;
  	env = dbp->env;
  	mpf = dbp->mpf;
  	cp = (HEAP_CURSOR *)dbc->internal;
  	putpage = FALSE;
  	ret = 0;
  
  	/*
  	 * We should have first page, locked already in cursor.  Get the
  	 * record id out of the cursor and set up local variables.
  	 */
  	DB_ASSERT(env, cp->page != NULL);
  	rid.pgno = cp->pgno;
  	rid.indx = cp->indx;
  	dpage = cp->page;
  	hdr = (HEAPSPLITHDR *)P_ENTRY(dbp, dpage, rid.indx);
  	DB_ASSERT(env, hdr->tsize != 0);
  	tlen = hdr->tsize;
  
  	/*
  	 * If we doing a partial retrieval, figure out how much we are
  	 * actually going to get.
  	 */
  	if (F_ISSET(dbt, DB_DBT_PARTIAL)) {
  		start = dbt->doff;
  		if (start > tlen)
  			needed = 0;
  		else if (dbt->dlen > tlen - start)
  			needed = tlen - start;
  		else
  			needed = dbt->dlen;
  	} else {
  		start = 0;
  		needed = tlen;
  	}
  
  	/*
  	 * If the caller has not requested any data, return success. This
  	 * "early-out" also avoids setting up the streaming optimization when
  	 * no page would be retrieved. If it were removed, the streaming code
  	 * should only initialize when needed is not 0.
  	 */
  	if (needed == 0) {
  		dbt->size = 0;
  		return (0);
  	}
  
  	/*
  	 * Check if the buffer is big enough; if it is not and we are
  	 * allowed to malloc space, then we'll malloc it.  If we are
  	 * not (DB_DBT_USERMEM), then we'll set the dbt and return
  	 * appropriately.
  	 */
  	if (F_ISSET(dbt, DB_DBT_USERCOPY))
  		goto skip_alloc;
  
  	/* Allocate any necessary memory. */
  	if (F_ISSET(dbt, DB_DBT_USERMEM)) {
  		if (needed > dbt->ulen) {
  			dbt->size = needed;
  			return (DB_BUFFER_SMALL);
  		}
  	} else if (F_ISSET(dbt, DB_DBT_MALLOC)) {
  		if ((ret = __os_umalloc(env, needed, &dbt->data)) != 0)
  			return (ret);
  	} else if (F_ISSET(dbt, DB_DBT_REALLOC)) {
  		if ((ret = __os_urealloc(env, needed, &dbt->data)) != 0)
  			return (ret);
  	} else if (bpsz != NULL && (*bpsz == 0 || *bpsz < needed)) {
  		if ((ret = __os_realloc(env, needed, bpp)) != 0)
  			return (ret);
  		*bpsz = needed;
  		dbt->data = *bpp;
  	} else if (bpp != NULL)
  		dbt->data = *bpp;
  	else {
  		DB_ASSERT(env,
  		    F_ISSET(dbt,
  		    DB_DBT_USERMEM | DB_DBT_MALLOC | DB_DBT_REALLOC) ||
  		    bpsz != NULL || bpp != NULL);
  		return (DB_BUFFER_SMALL);
  	}
  
  skip_alloc:
  	/*
  	 * Go through each of the pieces, copying the data on each one
  	 * into the buffer.  Never copy more than the total data length.
  	 * We are starting off with the page that is currently pointed to by
  	 * the cursor,
  	 */
  	curoff = 0;
  	dbt->size = needed;
  	for (p = dbt->data; needed > 0;) {
  		/* Check if we need any bytes from this page */
  		if (curoff + hdr->std_hdr.size >= start) {
  			bytes = hdr->std_hdr.size;
  			src = (u_int8_t *)hdr +
  			    P_TO_UINT16(sizeof(HEAPSPLITHDR));
  			if (start > curoff) {
  				src += start - curoff;
  				bytes -= start - curoff;
  			}
  			if (bytes > needed)
  				bytes = needed;
  			if (F_ISSET(dbt, DB_DBT_USERCOPY)) {
  				/*
  				 * The offset into the DBT is the total size
  				 * less the amount of data still needed.  Care
  				 * needs to be taken if doing a partial copy
  				 * beginning at an offset other than 0.
  				 */
  				if ((ret = env->dbt_usercopy(
  				    dbt, dbt->size - needed,
  				    src, bytes, DB_USERCOPY_SETDATA)) != 0) {
  					if (putpage)
  						(void)__memp_fput(
  						    mpf, dbc->thread_info,
  						    dpage, dbp->priority);
  
  					return (ret);
  				}
  			} else
  				memcpy(p, src, bytes);
  			p += bytes;
  			needed -= bytes;
  		}
  		curoff += hdr->std_hdr.size;
  
  		/* Find next record piece as long as it exists */
  		if (!F_ISSET((HEAPHDR *)hdr, HEAP_RECLAST)) {
  			rid.pgno = hdr->nextpg;
  			rid.indx = hdr->nextindx;
  
  			/*
  			 * First pass through here, we are using the
  			 * page pointed to by the cursor, and this page
  			 * will get put when the cursor is is closed.
  			 * Only pages specifically gotten in this loop
  			 * need to be put back.
  			 */
  			if (putpage) {
  				if ((ret = __memp_fput(mpf, dbc->thread_info,
  				    dpage, dbp->priority) ) != 0)
  					goto err;
  				dpage = NULL;
  				if ((ret = __TLPUT(dbc, data_lock)) != 0)
  					goto err;
  			}
  
  			if ((ret = __db_lget(dbc, 0, rid.pgno,
  			    DB_LOCK_READ, 0, &data_lock)) != 0)
  				goto err;
  			if ((ret = __memp_fget(mpf, &rid.pgno,
  			    dbc->thread_info, dbc->txn, 0, &dpage)) != 0)
  				goto err;
  			hdr = (HEAPSPLITHDR *)P_ENTRY(dbp, dpage, rid.indx);
  			putpage = TRUE;
  
  			/*
  			 * If we have the last piece of this record and we're
  			 * reading the entire record, then what we need should
  			 * equal what is remaining.
  			 */
  			if (F_ISSET((HEAPHDR *)hdr, HEAP_RECLAST) &&
  			    !F_ISSET(dbt, DB_DBT_PARTIAL) &&
  			    (hdr->std_hdr.size != needed)) {
  				__db_errx(env, DB_STR_A("1167",
  			     "Incorrect record size in header: %s: rid %lu.%lu",
  				    "%s %lu %lu"), dbc->dbp->fname,
  				    (u_long)(cp->pgno), (u_long)(cp->indx));
  				ret = __env_panic(env, DB_RUNRECOVERY);
  				goto err;
  			}
  		}
  	}
  
  err:	DB_ASSERT(dbp->env, ret != DB_PAGE_NOTFOUND);
  	if (putpage && dpage != NULL && (t_ret = __memp_fput(mpf,
  	    dbc->thread_info, dpage, dbp->priority)) != 0 && ret == 0)
  		ret = t_ret;
  	if ((t_ret = __TLPUT(dbc, data_lock)) != 0 && ret == 0)
  		ret = t_ret;
  
  	return (ret);
  }
  /*
   * __heapc_refresh --
   *      do the proper set up for cursor reuse.
   *
   * PUBLIC: int __heapc_refresh __P((DBC *));
   */
  int
  __heapc_refresh(dbc)
  	DBC *dbc;
  {
  	HEAP_CURSOR *cp;
  
  	cp = (HEAP_CURSOR *)dbc->internal;
  
  	LOCK_INIT(cp->lock);
  	cp->lock_mode = DB_LOCK_NG;
  	cp->flags = 0;
  
  	return (0);
  }
  