db2/btree/bt_put.c

   1 /*-
   2  * See the file LICENSE for redistribution information.
   3  *
   4  * Copyright (c) 1996, 1997
   5  *      Sleepycat Software.  All rights reserved.
   6  */
   7 /*
   8  * Copyright (c) 1990, 1993, 1994, 1995, 1996
   9  *      Keith Bostic.  All rights reserved.
  10  */
  11 /*
  12  * Copyright (c) 1990, 1993, 1994, 1995
  13  *      The Regents of the University of California.  All rights reserved.
  14  *
  15  * This code is derived from software contributed to Berkeley by
  16  * Mike Olson.
  17  *
  18  * Redistribution and use in source and binary forms, with or without
  19  * modification, are permitted provided that the following conditions
  20  * are met:
  21  * 1. Redistributions of source code must retain the above copyright
  22  *    notice, this list of conditions and the following disclaimer.
  23  * 2. Redistributions in binary form must reproduce the above copyright
  24  *    notice, this list of conditions and the following disclaimer in the
  25  *    documentation and/or other materials provided with the distribution.
  26  * 3. All advertising materials mentioning features or use of this software
  27  *    must display the following acknowledgement:
  28  *      This product includes software developed by the University of
  29  *      California, Berkeley and its contributors.
  30  * 4. Neither the name of the University nor the names of its contributors
  31  *    may be used to endorse or promote products derived from this software
  32  *    without specific prior written permission.
  33  *
  34  * THIS SOFTWARE IS PROVIDED BY THE REGENTS AND CONTRIBUTORS ``AS IS'' AND
  35  * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
  36  * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
  37  * ARE DISCLAIMED.  IN NO EVENT SHALL THE REGENTS OR CONTRIBUTORS BE LIABLE
  38  * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
  39  * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
  40  * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
  41  * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
  42  * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
  43  * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
  44  * SUCH DAMAGE.
  45  */
  46
  47 #include "config.h"
  48
  49 #ifndef lint
  50 static const char sccsid[] = "@(#)bt_put.c      10.31 (Sleepycat) 10/26/97";
  51 #endif /* not lint */
  52
  53 #ifndef NO_SYSTEM_INCLUDES
  54 #include <sys/types.h>
  55
  56 #include <errno.h>
  57 #include <stdio.h>
  58 #include <stdlib.h>
  59 #include <string.h>
  60 #endif
  61
  62 #include "db_int.h"
  63 #include "db_page.h"
  64 #include "btree.h"
  65
  66 static int __bam_fixed __P((BTREE *, DBT *));
  67 static int __bam_lookup __P((DB *, DBT *, int *));
  68 static int __bam_ndup __P((DB *, PAGE *, u_int32_t));
  69 static int __bam_ovput __P((DB *, PAGE *, u_int32_t, DBT *));
  70 static int __bam_partial __P((DB *, DBT *, PAGE *, u_int32_t, u_int32_t));
  71 static u_int32_t
  72            __bam_partsize __P((DB *, DBT *, PAGE *, u_int32_t));
  73
  74 /*
  75  * __bam_put --
  76  *      Add a new key/data pair or replace an existing pair (btree).
  77  *
  78  * PUBLIC: int __bam_put __P((DB *, DB_TXN *, DBT *, DBT *, int));
  79  */
  80 int
  81 __bam_put(argdbp, txn, key, data, flags)
  82         DB *argdbp;
  83         DB_TXN *txn;
  84         DBT *key, *data;
  85         int flags;
  86 {
  87         BTREE *t;
  88         CURSOR c;
  89         DB *dbp;
  90         PAGE *h;
  91         db_indx_t indx;
  92         int exact, iflags, newkey, replace, ret, stack;
  93
  94         DEBUG_LWRITE(argdbp, txn, "bam_put", key, data, flags);
  95
  96         /* Check flags. */
  97         if ((ret = __db_putchk(argdbp, key, data, flags,
  98             F_ISSET(argdbp, DB_AM_RDONLY), F_ISSET(argdbp, DB_AM_DUP))) != 0)
  99                 return (ret);
 100
 101         GETHANDLE(argdbp, txn, &dbp, ret);
 102         t = dbp->internal;
 103
 104 retry:  /*
 105          * Find the location at which to insert.  The call to __bam_lookup
 106          * leaves the returned page pinned.
 107          */
 108         if ((ret = __bam_lookup(dbp, key, &exact)) != 0) {
 109                 PUTHANDLE(dbp);
 110                 return (ret);
 111         }
 112         h = t->bt_csp->page;
 113         indx = t->bt_csp->indx;
 114         stack = 1;
 115
 116         /*
 117          * If an identical key is already in the tree, and DB_NOOVERWRITE is
 118          * set, an error is returned.  If an identical key is already in the
 119          * tree and DB_NOOVERWRITE is not set, the key is either added (when
 120          * duplicates are permitted) or an error is returned.  The exception
 121          * is when the item located is referenced by a cursor and marked for
 122          * deletion, in which case we permit the overwrite and flag the cursor.
 123          */
 124         replace = 0;
 125         if (exact && flags == DB_NOOVERWRITE) {
 126                 if (!B_DISSET(GET_BKEYDATA(h, indx + O_INDX)->type)) {
 127                         ret = DB_KEYEXIST;
 128                         goto err;
 129                 }
 130                 replace = 1;
 131                 __bam_ca_replace(dbp, h->pgno, indx, REPLACE_SETUP);
 132         }
 133
 134         /*
 135          * If we're inserting into the first or last page of the tree,
 136          * remember where we did it so we can do fast lookup next time.
 137          *
 138          * XXX
 139          * Does reverse order still work (did it ever!?!?)
 140          */
 141         t->bt_lpgno =
 142             h->next_pgno == PGNO_INVALID || h->prev_pgno == PGNO_INVALID ?
 143             h->pgno : PGNO_INVALID;
 144
 145         /*
 146          * Select the arguments for __bam_iitem() and do the insert.  If the
 147          * key is an exact match, we're either adding a new duplicate at the
 148          * end of the duplicate set, or we're replacing the data item with a
 149          * new data item.  If the key isn't an exact match, we're inserting
 150          * a new key/data pair, before the search location.
 151          */
 152         newkey = dbp->type == DB_BTREE && !exact;
 153         if (exact) {
 154                 if (F_ISSET(dbp, DB_AM_DUP)) {
 155                         /*
 156                          * Make sure that we're not looking at a page of
 157                          * duplicates -- if so, move to the last entry on
 158                          * that page.
 159                          */
 160                         c.page = h;
 161                         c.pgno = h->pgno;
 162                         c.indx = indx;
 163                         c.dpgno = PGNO_INVALID;
 164                         c.dindx = 0;
 165                         if ((ret =
 166                             __bam_ovfl_chk(dbp, &c, indx + O_INDX, 1)) != 0)
 167                                 goto err;
 168                         if (c.dpgno != PGNO_INVALID) {
 169                                 /*
 170                                  * XXX
 171                                  * The __bam_ovfl_chk() routine memp_fput() the
 172                                  * current page and acquired a new one, but did
 173                                  * not do anything about the lock we're holding.
 174                                  */
 175                                 t->bt_csp->page = h = c.page;
 176                                 indx = c.dindx;
 177                         }
 178                         iflags = DB_AFTER;
 179                 } else
 180                         iflags = DB_CURRENT;
 181         } else
 182                 iflags = DB_BEFORE;
 183
 184         /*
 185          * The pages we're using may be modified by __bam_iitem(), so make
 186          * sure we reset the stack.
 187          */
 188         ret = __bam_iitem(dbp,
 189             &h, &indx, key, data, iflags, newkey ? BI_NEWKEY : 0);
 190         t->bt_csp->page = h;
 191         t->bt_csp->indx = indx;
 192
 193         switch (ret) {
 194         case 0:
 195                 /*
 196                  * Done.  Clean up the cursor, and, if we're doing record
 197                  * numbers, adjust the internal page counts.
 198                  */
 199                 if (replace)
 200                         __bam_ca_replace(dbp, h->pgno, indx, REPLACE_SUCCESS);
 201
 202                 if (!replace && F_ISSET(dbp, DB_BT_RECNUM))
 203                         ret = __bam_adjust(dbp, t, 1);
 204                 break;
 205         case DB_NEEDSPLIT:
 206                 /*
 207                  * We have to split the page.  Back out the cursor setup,
 208                  * discard the stack of pages, and do the split.
 209                  */
 210                 if (replace) {
 211                         replace = 0;
 212                         __bam_ca_replace(dbp, h->pgno, indx, REPLACE_FAILED);
 213                 }
 214
 215                 (void)__bam_stkrel(dbp);
 216                 stack = 0;
 217
 218                 if ((ret = __bam_split(dbp, key)) != 0)
 219                         break;
 220
 221                 goto retry;
 222                 /* NOTREACHED */
 223         default:
 224                 if (replace)
 225                         __bam_ca_replace(dbp, h->pgno, indx, REPLACE_FAILED);
 226                 break;
 227         }
 228
 229 err:    if (stack)
 230                 (void)__bam_stkrel(dbp);
 231
 232         PUTHANDLE(dbp);
 233         return (ret);
 234 }
 235
 236 /*
 237  * __bam_lookup --
 238  *      Find the right location in the tree for the key.
 239  */
 240 static int
 241 __bam_lookup(dbp, key, exactp)
 242         DB *dbp;
 243         DBT *key;
 244         int *exactp;
 245 {
 246         BTREE *t;
 247         DB_LOCK lock;
 248         EPG e;
 249         PAGE *h;
 250         db_indx_t indx;
 251         int cmp, ret;
 252
 253         t = dbp->internal;
 254         h = NULL;
 255
 256         /*
 257          * Record numbers can't be fast-tracked, we have to lock the entire
 258          * tree.
 259          */
 260         if (F_ISSET(dbp, DB_BT_RECNUM))
 261                 goto slow;
 262
 263         /* Check to see if we've been seeing sorted input. */
 264         if (t->bt_lpgno == PGNO_INVALID)
 265                 goto slow;
 266
 267         /*
 268          * Retrieve the page on which we did the last insert.  It's okay if
 269          * it doesn't exist, or if it's not the page type we expect, it just
 270          * means that the world changed.
 271          */
 272         if (__bam_lget(dbp, 0, t->bt_lpgno, DB_LOCK_WRITE, &lock))
 273                 goto miss;
 274         if (__bam_pget(dbp, &h, &t->bt_lpgno, 0)) {
 275                 (void)__BT_LPUT(dbp, lock);
 276                 goto miss;
 277         }
 278         if (TYPE(h) != P_LBTREE)
 279                 goto miss;
 280         if (NUM_ENT(h) == 0)
 281                 goto miss;
 282
 283         /*
 284          * We have to be at the end or beginning of the tree to know that
 285          * we're inserting in a sort order.  If that's the case and we're
 286          * in the right order in comparison to the first/last key/data pair,
 287          * we have the right position.
 288          */
 289         if (h->next_pgno == PGNO_INVALID) {
 290                 e.page = h;
 291                 e.indx = NUM_ENT(h) - P_INDX;
 292                 if ((cmp = __bam_cmp(dbp, key, &e)) >= 0) {
 293                         if (cmp > 0)
 294                                 e.indx += P_INDX;
 295                         goto fast;
 296                 }
 297         }
 298         if (h->prev_pgno == PGNO_INVALID) {
 299                 e.page = h;
 300                 e.indx = 0;
 301                 if ((cmp = __bam_cmp(dbp, key, &e)) <= 0) {
 302                         /*
 303                          * We're doing a put, so we want to insert as the last
 304                          * of any set of duplicates.
 305                          */
 306                         if (cmp == 0) {
 307                                 for (indx = 0;
 308                                     indx < (db_indx_t)(NUM_ENT(h) - P_INDX) &&
 309                                     h->inp[indx] == h->inp[indx + P_INDX];
 310                                     indx += P_INDX);
 311                                 e.indx = indx;
 312                         }
 313                         goto fast;
 314                 }
 315         }
 316         goto miss;
 317
 318         /* Set the exact match flag in case we've already inserted this key. */
 319 fast:   *exactp = cmp == 0;
 320
 321         /* Enter the entry in the stack. */
 322         BT_STK_CLR(t);
 323         BT_STK_ENTER(t, e.page, e.indx, lock, ret);
 324         if (ret != 0)
 325                 return (ret);
 326
 327         ++t->lstat.bt_cache_hit;
 328         return (0);
 329
 330 miss:   ++t->lstat.bt_cache_miss;
 331         if (h != NULL) {
 332                 (void)memp_fput(dbp->mpf, h, 0);
 333                 (void)__BT_LPUT(dbp, lock);
 334         }
 335
 336 slow:   return (__bam_search(dbp, key, S_INSERT, 1, NULL, exactp));
 337 }
 338
 339 /*
 340  * __bam_iitem --
 341  *      Insert an item into the tree.
 342  *
 343  * PUBLIC: int __bam_iitem __P((DB *,
 344  * PUBLIC:    PAGE **, db_indx_t *, DBT *, DBT *, int, int));
 345  */
 346 int
 347 __bam_iitem(dbp, hp, indxp, key, data, op, flags)
 348         DB *dbp;
 349         PAGE **hp;
 350         db_indx_t *indxp;
 351         DBT *key, *data;
 352         int op, flags;
 353 {
 354         BTREE *t;
 355         BKEYDATA *bk;
 356         DBT tdbt;
 357         PAGE *h;
 358         db_indx_t indx;
 359         u_int32_t data_size, have_bytes, need_bytes, needed;
 360         int bigkey, bigdata, dupadjust, replace, ret;
 361
 362         t = dbp->internal;
 363         h = *hp;
 364         indx = *indxp;
 365
 366         bk = NULL;                      /* XXX: Shut the compiler up. */
 367         dupadjust = replace = 0;
 368
 369         /*
 370          * If it's a page of duplicates, call the common code to do the work.
 371          *
 372          * !!!
 373          * Here's where the hp and indxp are important.  The duplicate code
 374          * may decide to rework/rearrange the pages and indices we're using,
 375          * so the caller must understand that the page stack may change.
 376          */
 377         if (TYPE(h) == P_DUPLICATE) {
 378                 /* Adjust the index for the new item if it's a DB_AFTER op. */
 379                 if (op == DB_AFTER)
 380                         ++*indxp;
 381
 382                 /* Remove the current item if it's a DB_CURRENT op. */
 383                 if (op == DB_CURRENT && (ret = __db_ditem(dbp, *hp, *indxp,
 384                     BKEYDATA_SIZE(GET_BKEYDATA(*hp, *indxp)->len))) != 0)
 385                         return (ret);
 386
 387                 /* Put the new/replacement item onto the page. */
 388                 return (__db_dput(dbp, data, hp, indxp, __bam_new));
 389         }
 390
 391         /* Handle fixed-length records: build the real record. */
 392         if (F_ISSET(dbp, DB_RE_FIXEDLEN) && data->size != t->bt_recno->re_len) {
 393                 tdbt = *data;
 394                 if ((ret = __bam_fixed(t, &tdbt)) != 0)
 395                         return (ret);
 396                 data = &tdbt;
 397         }
 398
 399         /*
 400          * Figure out how much space the data will take, including if it's a
 401          * partial record.  If either of the key or data items won't fit on
 402          * a page, we'll have to store them on overflow pages.
 403          */
 404         bigkey = LF_ISSET(BI_NEWKEY) && key->size > t->bt_ovflsize;
 405         data_size = F_ISSET(data, DB_DBT_PARTIAL) ?
 406             __bam_partsize(dbp, data, h, indx) : data->size;
 407         bigdata = data_size > t->bt_ovflsize;
 408
 409         needed = 0;
 410         if (LF_ISSET(BI_NEWKEY)) {
 411                 /* If BI_NEWKEY is set we're adding a new key and data pair. */
 412                 if (bigkey)
 413                         needed += BOVERFLOW_PSIZE;
 414                 else
 415                         needed += BKEYDATA_PSIZE(key->size);
 416                 if (bigdata)
 417                         needed += BOVERFLOW_PSIZE;
 418                 else
 419                         needed += BKEYDATA_PSIZE(data_size);
 420         } else {
 421                 /*
 422                  * We're either overwriting the data item of a key/data pair
 423                  * or we're adding the data item only, i.e. a new duplicate.
 424                  */
 425                 if (op == DB_CURRENT) {
 426                         bk = GET_BKEYDATA(h,
 427                             indx + (TYPE(h) == P_LBTREE ? O_INDX : 0));
 428                         if (B_TYPE(bk->type) == B_OVERFLOW)
 429                                 have_bytes = BOVERFLOW_PSIZE;
 430                         else
 431                                 have_bytes = BKEYDATA_PSIZE(bk->len);
 432                         need_bytes = 0;
 433                 } else {
 434                         have_bytes = 0;
 435                         need_bytes = sizeof(db_indx_t);
 436                 }
 437                 if (bigdata)
 438                         need_bytes += BOVERFLOW_PSIZE;
 439                 else
 440                         need_bytes += BKEYDATA_PSIZE(data_size);
 441
 442                 if (have_bytes < need_bytes)
 443                         needed += need_bytes - have_bytes;
 444         }
 445
 446         /*
 447          * If there's not enough room, or the user has put a ceiling on the
 448          * number of keys permitted in the page, split the page.
 449          *
 450          * XXX
 451          * The t->bt_maxkey test here may be insufficient -- do we have to
 452          * check in the btree split code, so we don't undo it there!?!?
 453          */
 454         if (P_FREESPACE(h) < needed ||
 455             (t->bt_maxkey != 0 && NUM_ENT(h) > t->bt_maxkey))
 456                 return (DB_NEEDSPLIT);
 457
 458         /* Handle partial puts: build the real record. */
 459         if (F_ISSET(data, DB_DBT_PARTIAL)) {
 460                 tdbt = *data;
 461                 if ((ret = __bam_partial(dbp, &tdbt, h, indx, data_size)) != 0)
 462                         return (ret);
 463                 data = &tdbt;
 464         }
 465
 466         /*
 467          * The code breaks it up into six cases:
 468          *
 469          * 1. Append a new key/data pair.
 470          * 2. Insert a new key/data pair.
 471          * 3. Append a new data item (a new duplicate).
 472          * 4. Insert a new data item (a new duplicate).
 473          * 5. Overflow item: delete and re-add the data item.
 474          * 6. Replace the data item.
 475          */
 476         if (LF_ISSET(BI_NEWKEY)) {
 477                 switch (op) {
 478                 case DB_AFTER:          /* 1. Append a new key/data pair. */
 479                         indx += 2;
 480                         *indxp += 2;
 481                         break;
 482                 case DB_BEFORE:         /* 2. Insert a new key/data pair. */
 483                         break;
 484                 default:
 485                         abort();
 486                 }
 487
 488                 /* Add the key. */
 489                 if (bigkey) {
 490                         if ((ret = __bam_ovput(dbp, h, indx, key)) != 0)
 491                                 return (ret);
 492                 } else
 493                         if ((ret = __db_pitem(dbp, h, indx,
 494                             BKEYDATA_SIZE(key->size), NULL, key)) != 0)
 495                                 return (ret);
 496                 ++indx;
 497         } else {
 498                 switch (op) {
 499                 case DB_AFTER:          /* 3. Append a new data item. */
 500                         if (TYPE(h) == P_LBTREE) {
 501                                 /*
 502                                  * Adjust the cursor and copy in the key for
 503                                  * the duplicate.
 504                                  */
 505                                 if ((ret = __bam_adjindx(dbp,
 506                                     h, indx + P_INDX, indx, 1)) != 0)
 507                                         return (ret);
 508
 509                                 indx += 3;
 510                                 dupadjust = 1;
 511
 512                                 *indxp += 2;
 513                         } else {
 514                                 ++indx;
 515                                 __bam_ca_di(dbp, h->pgno, indx, 1);
 516
 517                                 *indxp += 1;
 518                         }
 519                         break;
 520                 case DB_BEFORE:         /* 4. Insert a new data item. */
 521                         if (TYPE(h) == P_LBTREE) {
 522                                 /*
 523                                  * Adjust the cursor and copy in the key for
 524                                  * the duplicate.
 525                                  */
 526                                 if ((ret =
 527                                     __bam_adjindx(dbp, h, indx, indx, 1)) != 0)
 528                                         return (ret);
 529
 530                                 ++indx;
 531                                 dupadjust = 1;
 532                         } else
 533                                 __bam_ca_di(dbp, h->pgno, indx, 1);
 534                         break;
 535                 case DB_CURRENT:
 536                         if (TYPE(h) == P_LBTREE)
 537                                 ++indx;
 538
 539                         /*
 540                          * 5. Delete/re-add the data item.
 541                          *
 542                          * If we're dealing with offpage items, we have to
 543                          * delete and then re-add the item.
 544                          */
 545                         if (bigdata || B_TYPE(bk->type) == B_OVERFLOW) {
 546                                 if ((ret = __bam_ditem(dbp, h, indx)) != 0)
 547                                         return (ret);
 548                                 break;
 549                         }
 550
 551                         /* 6. Replace the data item. */
 552                         replace = 1;
 553                         break;
 554                 default:
 555                         abort();
 556                 }
 557         }
 558
 559         /* Add the data. */
 560         if (bigdata) {
 561                 if ((ret = __bam_ovput(dbp, h, indx, data)) != 0)
 562                         return (ret);
 563         } else {
 564                 BKEYDATA __bk;
 565                 DBT __hdr;
 566
 567                 if (LF_ISSET(BI_DELETED)) {
 568                         B_TSET(__bk.type, B_KEYDATA, 1);
 569                         __bk.len = data->size;
 570                         __hdr.data = &__bk;
 571                         __hdr.size = SSZA(BKEYDATA, data);
 572                         ret = __db_pitem(dbp, h, indx,
 573                             BKEYDATA_SIZE(data->size), &__hdr, data);
 574                 } else if (replace)
 575                         ret = __bam_ritem(dbp, h, indx, data);
 576                 else
 577                         ret = __db_pitem(dbp, h, indx,
 578                             BKEYDATA_SIZE(data->size), NULL, data);
 579                 if (ret != 0)
 580                         return (ret);
 581         }
 582
 583         ++t->lstat.bt_added;
 584
 585         ret = memp_fset(dbp->mpf, h, DB_MPOOL_DIRTY);
 586
 587         /*
 588          * If the page is at least 50% full, and we added a duplicate, see if
 589          * that set of duplicates takes up at least 25% of the space.  If it
 590          * does, move it off onto its own page.
 591          */
 592         if (dupadjust && P_FREESPACE(h) <= dbp->pgsize / 2) {
 593                 --indx;
 594                 if ((ret = __bam_ndup(dbp, h, indx)) != 0)
 595                         return (ret);
 596         }
 597
 598         if (t->bt_recno != NULL)
 599                 F_SET(t->bt_recno, RECNO_MODIFIED);
 600
 601         return (ret);
 602 }
 603
 604 /*
 605  * __bam_partsize --
 606  *      Figure out how much space a partial data item is in total.
 607  */
 608 static u_int32_t
 609 __bam_partsize(dbp, data, h, indx)
 610         DB *dbp;
 611         DBT *data;
 612         PAGE *h;
 613         u_int32_t indx;
 614 {
 615         BKEYDATA *bk;
 616         u_int32_t nbytes;
 617
 618         /*
 619          * Figure out how much total space we'll need.  If the record doesn't
 620          * already exist, it's simply the data we're provided.
 621          */
 622         if (indx >= NUM_ENT(h))
 623                 return (data->doff + data->size);
 624
 625         /*
 626          * Otherwise, it's the data provided plus any already existing data
 627          * that we're not replacing.
 628          */
 629         bk = GET_BKEYDATA(h, indx + (TYPE(h) == P_LBTREE ? O_INDX : 0));
 630         nbytes =
 631             B_TYPE(bk->type) == B_OVERFLOW ? ((BOVERFLOW *)bk)->tlen : bk->len;
 632
 633         /*
 634          * There are really two cases here:
 635          *
 636          * Case 1: We are replacing some bytes that do not exist (i.e., they
 637          * are past the end of the record).  In this case the number of bytes
 638          * we are replacing is irrelevant and all we care about is how many
 639          * bytes we are going to add from offset.  So, the new record length
 640          * is going to be the size of the new bytes (size) plus wherever those
 641          * new bytes begin (doff).
 642          *
 643          * Case 2: All the bytes we are replacing exist.  Therefore, the new
 644          * size is the oldsize (nbytes) minus the bytes we are replacing (dlen)
 645          * plus the bytes we are adding (size).
 646          */
 647         if (nbytes < data->doff + data->dlen)           /* Case 1 */
 648                 return (data->doff + data->size);
 649
 650         return (nbytes + data->size - data->dlen);      /* Case 2 */
 651 }
 652
 653 /*
 654  * OVPUT --
 655  *      Copy an overflow item onto a page.
 656  */
 657 #undef  OVPUT
 658 #define OVPUT(h, indx, bo) do {                                         \
 659         DBT __hdr;                                                      \
 660         memset(&__hdr, 0, sizeof(__hdr));                               \
 661         __hdr.data = &bo;                                               \
 662         __hdr.size = BOVERFLOW_SIZE;                                    \
 663         if ((ret = __db_pitem(dbp,                                      \
 664             h, indx, BOVERFLOW_SIZE, &__hdr, NULL)) != 0)               \
 665                 return (ret);                                           \
 666 } while (0)
 667
 668 /*
 669  * __bam_ovput --
 670  *      Build an overflow item and put it on the page.
 671  */
 672 static int
 673 __bam_ovput(dbp, h, indx, item)
 674         DB *dbp;
 675         PAGE *h;
 676         u_int32_t indx;
 677         DBT *item;
 678 {
 679         BOVERFLOW bo;
 680         int ret;
 681
 682         B_TSET(bo.type, B_OVERFLOW, 0);
 683         bo.tlen = item->size;
 684         if ((ret = __db_poff(dbp, item, &bo.pgno, __bam_new)) != 0)
 685                 return (ret);
 686
 687         OVPUT(h, indx, bo);
 688
 689         return (0);
 690 }
 691
 692 /*
 693  * __bam_ritem --
 694  *      Replace an item on a page.
 695  *
 696  * PUBLIC: int __bam_ritem __P((DB *, PAGE *, u_int32_t, DBT *));
 697  */
 698 int
 699 __bam_ritem(dbp, h, indx, data)
 700         DB *dbp;
 701         PAGE *h;
 702         u_int32_t indx;
 703         DBT *data;
 704 {
 705         BKEYDATA *bk;
 706         DBT orig, repl;
 707         db_indx_t lo, ln, min, off, prefix, suffix;
 708         int32_t nbytes;
 709         int cnt, ret;
 710         u_int8_t *p, *t;
 711
 712         /*
 713          * Replace a single item onto a page.  The logic figuring out where
 714          * to insert and whether it fits is handled in the caller.  All we do
 715          * here is manage the page shuffling.
 716          */
 717         bk = GET_BKEYDATA(h, indx);
 718
 719         /* Log the change. */
 720         if (DB_LOGGING(dbp)) {
 721                 /*
 722                  * We might as well check to see if the two data items share
 723                  * a common prefix and suffix -- it can save us a lot of log
 724                  * message if they're large.
 725                  */
 726                 min = data->size < bk->len ? data->size : bk->len;
 727                 for (prefix = 0,
 728                     p = bk->data, t = data->data;
 729                     prefix < min && *p == *t; ++prefix, ++p, ++t)
 730                         ;
 731
 732                 min -= prefix;
 733                 for (suffix = 0,
 734                     p = (u_int8_t *)bk->data + bk->len - 1,
 735                     t = (u_int8_t *)data->data + data->size - 1;
 736                     suffix < min && *p == *t; ++suffix, --p, --t)
 737                         ;
 738
 739                 /* We only log the parts of the keys that have changed. */
 740                 orig.data = (u_int8_t *)bk->data + prefix;
 741                 orig.size = bk->len - (prefix + suffix);
 742                 repl.data = (u_int8_t *)data->data + prefix;
 743                 repl.size = data->size - (prefix + suffix);
 744                 if ((ret = __bam_repl_log(dbp->dbenv->lg_info, dbp->txn,
 745                     &LSN(h), 0, dbp->log_fileid, PGNO(h), &LSN(h),
 746                     (u_int32_t)indx, (u_int32_t)B_DISSET(bk->type),
 747                     &orig, &repl, (u_int32_t)prefix, (u_int32_t)suffix)) != 0)
 748                         return (ret);
 749         }
 750
 751         /*
 752          * Set references to the first in-use byte on the page and the
 753          * first byte of the item being replaced.
 754          */
 755         p = (u_int8_t *)h + HOFFSET(h);
 756         t = (u_int8_t *)bk;
 757
 758         /*
 759          * If the entry is growing in size, shift the beginning of the data
 760          * part of the page down.  If the entry is shrinking in size, shift
 761          * the beginning of the data part of the page up.  Use memmove(3),
 762          * the regions overlap.
 763          */
 764         lo = BKEYDATA_SIZE(bk->len);
 765         ln = BKEYDATA_SIZE(data->size);
 766         if (lo != ln) {
 767                 nbytes = lo - ln;               /* Signed difference. */
 768                 if (p == t)                     /* First index is fast. */
 769                         h->inp[indx] += nbytes;
 770                 else {                          /* Else, shift the page. */
 771                         memmove(p + nbytes, p, t - p);
 772
 773                         /* Adjust the indices' offsets. */
 774                         off = h->inp[indx];
 775                         for (cnt = 0; cnt < NUM_ENT(h); ++cnt)
 776                                 if (h->inp[cnt] <= off)
 777                                         h->inp[cnt] += nbytes;
 778                 }
 779
 780                 /* Clean up the page and adjust the item's reference. */
 781                 HOFFSET(h) += nbytes;
 782                 t += nbytes;
 783         }
 784
 785         /* Copy the new item onto the page. */
 786         bk = (BKEYDATA *)t;
 787         B_TSET(bk->type, B_KEYDATA, 0);
 788         bk->len = data->size;
 789         memcpy(bk->data, data->data, data->size);
 790
 791         return (0);
 792 }
 793
 794 /*
 795  * __bam_ndup --
 796  *      Check to see if the duplicate set at indx should have its own page.
 797  *      If it should, create it.
 798  */
 799 static int
 800 __bam_ndup(dbp, h, indx)
 801         DB *dbp;
 802         PAGE *h;
 803         u_int32_t indx;
 804 {
 805         BKEYDATA *bk;
 806         BOVERFLOW bo;
 807         DBT hdr;
 808         PAGE *cp;
 809         db_indx_t cnt, cpindx, first, sz;
 810         int ret;
 811
 812         while (indx > 0 && h->inp[indx] == h->inp[indx - P_INDX])
 813                 indx -= P_INDX;
 814         for (cnt = 0, sz = 0, first = indx;; ++cnt, indx += P_INDX) {
 815                 if (indx >= NUM_ENT(h) || h->inp[first] != h->inp[indx])
 816                         break;
 817                 bk = GET_BKEYDATA(h, indx);
 818                 sz += B_TYPE(bk->type) == B_KEYDATA ?
 819                     BKEYDATA_PSIZE(bk->len) : BOVERFLOW_PSIZE;
 820                 bk = GET_BKEYDATA(h, indx + O_INDX);
 821                 sz += B_TYPE(bk->type) == B_KEYDATA ?
 822                     BKEYDATA_PSIZE(bk->len) : BOVERFLOW_PSIZE;
 823         }
 824
 825         /*
 826          * If this set of duplicates is using more than 25% of the page, move
 827          * them off.  The choice of 25% is a WAG, but it has to be small enough
 828          * that we can always split regardless of the presence of duplicates.
 829          */
 830         if (sz < dbp->pgsize / 4)
 831                 return (0);
 832
 833         /* Get a new page. */
 834         if ((ret = __bam_new(dbp, P_DUPLICATE, &cp)) != 0)
 835                 return (ret);
 836
 837         /*
 838          * Move this set of duplicates off the page.  First points to the first
 839          * key of the first duplicate key/data pair, cnt is the number of pairs
 840          * we're dealing with.
 841          */
 842         memset(&hdr, 0, sizeof(hdr));
 843         for (indx = first + O_INDX, cpindx = 0;; ++cpindx) {
 844                 /* Copy the entry to the new page. */
 845                 bk = GET_BKEYDATA(h, indx);
 846                 hdr.data = bk;
 847                 hdr.size = B_TYPE(bk->type) == B_KEYDATA ?
 848                     BKEYDATA_SIZE(bk->len) : BOVERFLOW_SIZE;
 849                 if ((ret =
 850                     __db_pitem(dbp, cp, cpindx, hdr.size, &hdr, NULL)) != 0)
 851                         goto err;
 852
 853                 /*
 854                  * Move cursors referencing the old entry to the new entry.
 855                  * Done after the page put because __db_pitem() adjusts
 856                  * cursors on the new page, and before the delete because
 857                  * __db_ditem adjusts cursors on the old page.
 858                  */
 859                 __bam_ca_dup(dbp,
 860                     PGNO(h), first, indx - O_INDX, PGNO(cp), cpindx);
 861
 862                 /* Delete the data item. */
 863                 if ((ret = __db_ditem(dbp, h, indx, hdr.size)) != 0)
 864                         goto err;
 865
 866                 /* Delete all but the first reference to the key. */
 867                 if (--cnt == 0)
 868                         break;
 869                 if ((ret = __bam_adjindx(dbp, h, indx, first, 0)) != 0)
 870                         goto err;
 871         }
 872
 873         /* Put in a new data item that points to the duplicates page. */
 874         B_TSET(bo.type, B_DUPLICATE, 0);
 875         bo.pgno = cp->pgno;
 876         bo.tlen = 0;
 877
 878         OVPUT(h, indx, bo);
 879
 880         return (memp_fput(dbp->mpf, cp, DB_MPOOL_DIRTY));
 881
 882 err:    (void)__bam_free(dbp, cp);
 883         return (ret);
 884 }
 885
 886 /*
 887  * __bam_fixed --
 888  *      Build the real record for a fixed length put.
 889  */
 890 static int
 891 __bam_fixed(t, dbt)
 892         BTREE *t;
 893         DBT *dbt;
 894 {
 895         RECNO *rp;
 896
 897         rp = t->bt_recno;
 898
 899         /*
 900          * If database contains fixed-length records, and the record is long,
 901          * return EINVAL.
 902          */
 903         if (dbt->size > rp->re_len)
 904                 return (EINVAL);
 905
 906         /*
 907          * The caller checked to see if it was just right, so we know it's
 908          * short.  Pad it out.  We use the record data return memory, it's
 909          * only a short-term use.
 910          */
 911         if (t->bt_rdata.ulen < rp->re_len) {
 912                 t->bt_rdata.data = t->bt_rdata.data == NULL ?
 913                     (void *)__db_malloc(rp->re_len) :
 914                     (void *)__db_realloc(t->bt_rdata.data, rp->re_len);
 915                 if (t->bt_rdata.data == NULL) {
 916                         t->bt_rdata.ulen = 0;
 917                         return (ENOMEM);
 918                 }
 919                 t->bt_rdata.ulen = rp->re_len;
 920         }
 921         memcpy(t->bt_rdata.data, dbt->data, dbt->size);
 922         memset((u_int8_t *)t->bt_rdata.data + dbt->size,
 923             rp->re_pad, rp->re_len - dbt->size);
 924
 925         /*
 926          * Clean up our flags and other information just in case, and
 927          * change the caller's DBT to reference our created record.
 928          */
 929         t->bt_rdata.size = rp->re_len;
 930         t->bt_rdata.dlen = 0;
 931         t->bt_rdata.doff = 0;
 932         t->bt_rdata.flags = 0;
 933         *dbt = t->bt_rdata;
 934
 935         return (0);
 936 }
 937
 938 /*
 939  * __bam_partial --
 940  *      Build the real record for a partial put.
 941  */
 942 static int
 943 __bam_partial(dbp, dbt, h, indx, nbytes)
 944         DB *dbp;
 945         DBT *dbt;
 946         PAGE *h;
 947         u_int32_t indx, nbytes;
 948 {
 949         BTREE *t;
 950         BKEYDATA *bk, tbk;
 951         BOVERFLOW *bo;
 952         DBT copy;
 953         u_int32_t len, tlen;
 954         int ret;
 955         u_int8_t *p;
 956
 957         bo = NULL;                      /* XXX: Shut the compiler up. */
 958         t = dbp->internal;
 959
 960         /* We use the record data return memory, it's only a short-term use. */
 961         if (t->bt_rdata.ulen < nbytes) {
 962                 t->bt_rdata.data = t->bt_rdata.data == NULL ?
 963                     (void *)__db_malloc(nbytes) :
 964                     (void *)__db_realloc(t->bt_rdata.data, nbytes);
 965                 if (t->bt_rdata.data == NULL) {
 966                         t->bt_rdata.ulen = 0;
 967                         return (ENOMEM);
 968                 }
 969                 t->bt_rdata.ulen = nbytes;
 970         }
 971
 972         /* Find the current record. */
 973         if (indx < NUM_ENT(h)) {
 974                 bk = GET_BKEYDATA(h, indx + (TYPE(h) == P_LBTREE ? O_INDX : 0));
 975                 bo = (BOVERFLOW *)bk;
 976         } else {
 977                 bk = &tbk;
 978                 B_TSET(bk->type, B_KEYDATA, 0);
 979                 bk->len = 0;
 980         }
 981
 982         /* We use nul bytes for extending the record, get it over with. */
 983         memset(t->bt_rdata.data, 0, nbytes);
 984
 985         tlen = 0;
 986         if (B_TYPE(bk->type) == B_OVERFLOW) {
 987                 /* Take up to doff bytes from the record. */
 988                 memset(&copy, 0, sizeof(copy));
 989                 if ((ret = __db_goff(dbp, &copy, bo->tlen,
 990                     bo->pgno, &t->bt_rdata.data, &t->bt_rdata.ulen)) != 0)
 991                         return (ret);
 992                 tlen += dbt->doff;
 993
 994                 /*
 995                  * If the original record was larger than the offset:
 996                  *      If dlen > size, shift the remaining data down.
 997                  *      If dlen < size, shift the remaining data up.
 998                  * Use memmove(), the regions may overlap.
 999                  */
1000                 p = t->bt_rdata.data;
1001                 if (bo->tlen > dbt->doff)
1002                         if (dbt->dlen > dbt->size) {
1003                                 tlen += len = bo->tlen -
1004                                     dbt->doff - (dbt->dlen - dbt->size);
1005                                 memmove(p + dbt->doff + dbt->size,
1006                                     p + dbt->doff + dbt->dlen, len);
1007                         } else if (dbt->dlen < dbt->size) {
1008                                 tlen += len = bo->tlen -
1009                                     dbt->doff - (dbt->size - dbt->dlen);
1010                                 memmove(p + dbt->doff + dbt->dlen,
1011                                     p + dbt->doff + dbt->size, len);
1012                         } else
1013                                 tlen += bo->tlen - dbt->doff;
1014
1015                 /* Copy in the user's data. */
1016                 memcpy((u_int8_t *)t->bt_rdata.data + dbt->doff,
1017                     dbt->data, dbt->size);
1018                 tlen += dbt->size;
1019         } else {
1020                 /* Take up to doff bytes from the record. */
1021                 memcpy(t->bt_rdata.data,
1022                     bk->data, dbt->doff > bk->len ? bk->len : dbt->doff);
1023                 tlen += dbt->doff;
1024
1025                 /* Copy in the user's data. */
1026                 memcpy((u_int8_t *)t->bt_rdata.data +
1027                     dbt->doff, dbt->data, dbt->size);
1028                 tlen += dbt->size;
1029
1030                 /* Copy in any remaining data. */
1031                 len = dbt->doff + dbt->dlen;
1032                 if (bk->len > len) {
1033                         memcpy((u_int8_t *)t->bt_rdata.data + dbt->doff +
1034                             dbt->size, bk->data + len, bk->len - len);
1035                         tlen += bk->len - len;
1036                 }
1037         }
1038
1039         /* Set the DBT to reference our new record. */
1040         t->bt_rdata.size = tlen;
1041         t->bt_rdata.dlen = 0;
1042         t->bt_rdata.doff = 0;
1043         t->bt_rdata.flags = 0;
1044         *dbt = t->bt_rdata;
1045         return (0);
1046 }