db2/hash/hash_dup.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
   9  *      The Regents of the University of California.  All rights reserved.
  10  *
  11  * This code is derived from software contributed to Berkeley by
  12  * Margo Seltzer.
  13  *
  14  * Redistribution and use in source and binary forms, with or without
  15  * modification, are permitted provided that the following conditions
  16  * are met:
  17  * 1. Redistributions of source code must retain the above copyright
  18  *    notice, this list of conditions and the following disclaimer.
  19  * 2. Redistributions in binary form must reproduce the above copyright
  20  *    notice, this list of conditions and the following disclaimer in the
  21  *    documentation and/or other materials provided with the distribution.
  22  * 3. All advertising materials mentioning features or use of this software
  23  *    must display the following acknowledgement:
  24  *      This product includes software developed by the University of
  25  *      California, Berkeley and its contributors.
  26  * 4. Neither the name of the University nor the names of its contributors
  27  *    may be used to endorse or promote products derived from this software
  28  *    without specific prior written permission.
  29  *
  30  * THIS SOFTWARE IS PROVIDED BY THE REGENTS AND CONTRIBUTORS ``AS IS'' AND
  31  * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
  32  * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
  33  * ARE DISCLAIMED.  IN NO EVENT SHALL THE REGENTS OR CONTRIBUTORS BE LIABLE
  34  * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
  35  * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
  36  * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
  37  * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
  38  * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
  39  * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
  40  * SUCH DAMAGE.
  41  */
  42 #include "config.h"
  43
  44 #ifndef lint
  45 static const char sccsid[] = "@(#)hash_dup.c    10.8 (Sleepycat) 10/14/97";
  46 #endif /* not lint */
  47
  48 /*
  49  * PACKAGE:  hashing
  50  *
  51  * DESCRIPTION:
  52  *      Manipulation of duplicates for the hash package.
  53  *
  54  * ROUTINES:
  55  *
  56  * External
  57  *      __add_dup
  58  * Internal
  59  */
  60
  61 #ifndef NO_SYSTEM_INCLUDES
  62 #include <sys/types.h>
  63
  64 #include <stdio.h>
  65 #include <stdlib.h>
  66 #include <string.h>
  67 #include <unistd.h>
  68 #endif
  69
  70 #include "db_int.h"
  71 #include "db_page.h"
  72 #include "db_swap.h"
  73 #include "hash.h"
  74
  75 static int __ham_check_move __P((HTAB *, HASH_CURSOR *, int32_t));
  76 static int __ham_dup_convert __P((HTAB *, HASH_CURSOR *));
  77 static int __ham_make_dup __P((const DBT *, DBT *d, void **, u_int32_t *));
  78
  79 /*
  80  * Called from hash_access to add a duplicate key. nval is the new
  81  * value that we want to add.  The flags correspond to the flag values
  82  * to cursor_put indicating where to add the new element.
  83  * There are 4 cases.
  84  * Case 1: The existing duplicate set already resides on a separate page.
  85  *         We can use common code for this.
  86  * Case 2: The element is small enough to just be added to the existing set.
  87  * Case 3: The element is large enough to be a big item, so we're going to
  88  *         have to push the set onto a new page.
  89  * Case 4: The element is large enough to push the duplicate set onto a
  90  *         separate page.
  91  *
  92  * PUBLIC: int __ham_add_dup __P((HTAB *, HASH_CURSOR *, DBT *, int));
  93  */
  94 int
  95 __ham_add_dup(hashp, hcp, nval, flags)
  96         HTAB *hashp;
  97         HASH_CURSOR *hcp;
  98         DBT *nval;
  99         int flags;
 100 {
 101         DBT pval, tmp_val;
 102         u_int32_t del_len, new_size;
 103         int ret;
 104         u_int8_t *hk;
 105
 106         if (flags == DB_CURRENT && hcp->dpgno == PGNO_INVALID)
 107                 del_len = hcp->dup_len;
 108         else
 109                 del_len = 0;
 110
 111         if ((ret = __ham_check_move(hashp, hcp,
 112             (int32_t)DUP_SIZE(nval->size) - (int32_t)del_len)) != 0)
 113                 return (ret);
 114
 115         /*
 116          * Check if resulting duplicate set is going to need to go
 117          * onto a separate duplicate page.  If so, convert the
 118          * duplicate set and add the new one.  After conversion,
 119          * hcp->dndx is the first free ndx or the index of the
 120          * current pointer into the duplicate set.
 121          */
 122         hk = H_PAIRDATA(hcp->pagep, hcp->bndx);
 123         new_size = DUP_SIZE(nval->size) - del_len + LEN_HKEYDATA(hcp->pagep,
 124             hashp->hdr->pagesize, H_DATAINDEX(hcp->bndx));
 125
 126         /*
 127          * We convert to off-page duplicates if the item is a big item,
 128          * the addition of the new item will make the set large, or
 129          * if there isn't enough room on this page to add the next item.
 130          */
 131         if (HPAGE_PTYPE(hk) != H_OFFDUP &&
 132             (HPAGE_PTYPE(hk) == H_OFFPAGE || ISBIG(hashp, new_size) ||
 133             DUP_SIZE(nval->size) - del_len > P_FREESPACE(hcp->pagep))) {
 134
 135                 if ((ret = __ham_dup_convert(hashp, hcp)) != 0)
 136                         return (ret);
 137                 else
 138                         hk = H_PAIRDATA(hcp->pagep, hcp->bndx);
 139         }
 140
 141         /* There are two separate cases here: on page and off page. */
 142         if (HPAGE_PTYPE(hk) != H_OFFDUP) {
 143                 if (HPAGE_PTYPE(hk) != H_DUPLICATE) {
 144                         HPAGE_PTYPE(hk) = H_DUPLICATE;
 145                         pval.flags = 0;
 146                         pval.data = HKEYDATA_DATA(hk);
 147                         pval.size = LEN_HDATA(hcp->pagep, hashp->hdr->pagesize,
 148                             hcp->bndx);
 149                         if ((ret =
 150                             __ham_make_dup(&pval, &tmp_val, &hcp->big_data,
 151                             &hcp->big_datalen)) != 0 || (ret =
 152                             __ham_replpair(hashp, hcp, &tmp_val, 1)) != 0)
 153                                 return (ret);
 154                 }
 155
 156                 /* Now make the new entry a duplicate. */
 157                 if ((ret = __ham_make_dup(nval,
 158                     &tmp_val, &hcp->big_data, &hcp->big_datalen)) != 0)
 159                         return (ret);
 160
 161                 tmp_val.dlen = 0;
 162                 switch (flags) {                        /* On page. */
 163                 case DB_KEYFIRST:
 164                         tmp_val.doff = 0;
 165                         break;
 166                 case DB_KEYLAST:
 167                         tmp_val.doff = LEN_HDATA(hcp->pagep,
 168                             hashp->hdr->pagesize, hcp->bndx);
 169                         break;
 170                 case DB_CURRENT:
 171                         tmp_val.doff = hcp->dup_off;
 172                         tmp_val.dlen = DUP_SIZE(hcp->dup_len);
 173                         break;
 174                 case DB_BEFORE:
 175                         tmp_val.doff = hcp->dup_off;
 176                         break;
 177                 case DB_AFTER:
 178                         tmp_val.doff = hcp->dup_off + DUP_SIZE(hcp->dup_len);
 179                         break;
 180                 }
 181                 /* Add the duplicate. */
 182                 ret = __ham_replpair(hashp, hcp, &tmp_val, 0);
 183                 if (ret == 0)
 184                         ret = __ham_dirty_page(hashp, hcp->pagep);
 185                 __ham_c_update(hashp, hcp, hcp->pgno, tmp_val.size, 1, 1);
 186                 return (ret);
 187         }
 188
 189         /* If we get here, then we're on duplicate pages. */
 190         if (hcp->dpgno == PGNO_INVALID) {
 191                 memcpy(&hcp->dpgno, HOFFDUP_PGNO(hk), sizeof(db_pgno_t));
 192                 hcp->dndx = 0;
 193         }
 194
 195         switch (flags) {
 196         case DB_KEYFIRST:
 197                 /*
 198                  * The only way that we are already on a dup page is
 199                  * if we just converted the on-page representation.
 200                  * In that case, we've only got one page of duplicates.
 201                  */
 202                 if (hcp->dpagep == NULL && (ret =
 203                     __db_dend(hashp->dbp, hcp->dpgno, &hcp->dpagep)) != 0)
 204                         return (ret);
 205                 hcp->dndx = 0;
 206                 break;
 207         case DB_KEYLAST:
 208                 if (hcp->dpagep == NULL && (ret =
 209                     __db_dend(hashp->dbp, hcp->dpgno, &hcp->dpagep)) != 0)
 210                         return (ret);
 211                 hcp->dpgno = PGNO(hcp->dpagep);
 212                 hcp->dndx = NUM_ENT(hcp->dpagep);
 213                 break;
 214         case DB_CURRENT:
 215                 if ((ret = __db_ditem(hashp->dbp, hcp->dpagep, hcp->dndx,
 216                     BKEYDATA_SIZE(GET_BKEYDATA(hcp->dpagep, hcp->dndx)->len)))
 217                     != 0)
 218                         return (ret);
 219                 break;
 220         case DB_BEFORE: /* The default behavior is correct. */
 221                 break;
 222         case DB_AFTER:
 223                 hcp->dndx++;
 224                 break;
 225         }
 226
 227         ret = __db_dput(hashp->dbp,
 228             nval, &hcp->dpagep, &hcp->dndx, __ham_overflow_page);
 229         hcp->pgno = PGNO(hcp->pagep);
 230         __ham_c_update(hashp, hcp, hcp->pgno, nval->size, 1, 1);
 231         return (ret);
 232 }
 233
 234 /*
 235  * Convert an on-page set of duplicates to an offpage set of duplicates.
 236  */
 237 static int
 238 __ham_dup_convert(hashp, hcp)
 239         HTAB *hashp;
 240         HASH_CURSOR *hcp;
 241 {
 242         BOVERFLOW bo;
 243         DBT dbt;
 244         HOFFPAGE ho;
 245         db_indx_t dndx, len;
 246         int ret;
 247         u_int8_t *p, *pend;
 248
 249         /*
 250          * Create a new page for the duplicates.
 251          */
 252         if ((ret =
 253             __ham_overflow_page(hashp->dbp, P_DUPLICATE, &hcp->dpagep)) != 0)
 254                 return (ret);
 255         hcp->dpagep->type = P_DUPLICATE;
 256         hcp->dpgno = PGNO(hcp->dpagep);
 257
 258         /*
 259          * Now put the duplicates onto the new page.
 260          */
 261         dbt.flags = 0;
 262         switch (HPAGE_PTYPE(H_PAIRDATA(hcp->pagep, hcp->bndx))) {
 263         case H_KEYDATA:
 264                 /* Simple case, one key on page; move it to dup page. */
 265                 dndx = 0;
 266                 dbt.size =
 267                     LEN_HDATA(hcp->pagep, hashp->hdr->pagesize, hcp->bndx);
 268                 dbt.data = HKEYDATA_DATA(H_PAIRDATA(hcp->pagep, hcp->bndx));
 269                 ret = __db_pitem(hashp->dbp, hcp->dpagep,
 270                     (u_int32_t)dndx, BKEYDATA_SIZE(dbt.size), NULL, &dbt);
 271                 if (ret == 0)
 272                         __ham_dirty_page(hashp, hcp->dpagep);
 273                 break;
 274         case H_OFFPAGE:
 275                 /* Simple case, one key on page; move it to dup page. */
 276                 dndx = 0;
 277                 memcpy(&ho,
 278                     P_ENTRY(hcp->pagep, H_DATAINDEX(hcp->bndx)), HOFFPAGE_SIZE);
 279                 B_TSET(bo.type, ho.type, 0);
 280                 bo.pgno = ho.pgno;
 281                 bo.tlen = ho.tlen;
 282                 dbt.size = BOVERFLOW_SIZE;
 283                 dbt.data = &bo;
 284
 285                 ret = __db_pitem(hashp->dbp, hcp->dpagep,
 286                    (u_int32_t)dndx, dbt.size, &dbt, NULL);
 287                 if (ret == 0)
 288                         __ham_dirty_page(hashp, hcp->dpagep);
 289                 break;
 290         case H_DUPLICATE:
 291                 p = HKEYDATA_DATA(H_PAIRDATA(hcp->pagep, hcp->bndx));
 292                 pend = p +
 293                     LEN_HDATA(hcp->pagep, hashp->hdr->pagesize, hcp->bndx);
 294
 295                 for (dndx = 0; p < pend; dndx++) {
 296                         memcpy(&len, p, sizeof(db_indx_t));
 297                         dbt.size = len;
 298                         p += sizeof(db_indx_t);
 299                         dbt.data = p;
 300                         p += len + sizeof(db_indx_t);
 301                         ret = __db_dput(hashp->dbp, &dbt,
 302                             &hcp->dpagep, &dndx, __ham_overflow_page);
 303                         if (ret != 0)
 304                                 break;
 305                 }
 306                 break;
 307         default:
 308                 ret = __db_pgfmt(hashp->dbp, (u_long)hcp->pgno);
 309         }
 310         if (ret == 0) {
 311                 /*
 312                  * Now attach this to the source page in place of
 313                  * the old duplicate item.
 314                  */
 315                 __ham_move_offpage(hashp, hcp->pagep,
 316                     (u_int32_t)H_DATAINDEX(hcp->bndx), hcp->dpgno);
 317
 318                 /* Can probably just do a "put" here. */
 319                 ret = __ham_dirty_page(hashp, hcp->pagep);
 320         } else {
 321                 (void)__ham_del_page(hashp->dbp, hcp->dpagep);
 322                 hcp->dpagep = NULL;
 323         }
 324         return (ret);
 325 }
 326
 327 static int
 328 __ham_make_dup(notdup, dup, bufp, sizep)
 329         const DBT *notdup;
 330         DBT *dup;
 331         void **bufp;
 332         u_int32_t *sizep;
 333 {
 334         db_indx_t tsize, item_size;
 335         int ret;
 336         u_int8_t *p;
 337
 338         item_size = (db_indx_t)notdup->size;
 339         tsize = DUP_SIZE(item_size);
 340         if ((ret = __ham_init_dbt(dup, tsize, bufp, sizep)) != 0)
 341                 return (ret);
 342
 343         dup->dlen = 0;
 344         dup->flags = notdup->flags;
 345         F_SET(dup, DB_DBT_PARTIAL);
 346
 347         p = dup->data;
 348         memcpy(p, &item_size, sizeof(db_indx_t));
 349         p += sizeof(db_indx_t);
 350         memcpy(p, notdup->data, notdup->size);
 351         p += notdup->size;
 352         memcpy(p, &item_size, sizeof(db_indx_t));
 353
 354         dup->doff = 0;
 355         dup->dlen = notdup->size;
 356
 357         return (0);
 358 }
 359
 360 static int
 361 __ham_check_move(hashp, hcp, add_len)
 362         HTAB *hashp;
 363         HASH_CURSOR *hcp;
 364         int32_t add_len;
 365 {
 366         DBT k, d;
 367         DB_LSN new_lsn;
 368         PAGE *next_pagep;
 369         db_pgno_t next_pgno;
 370         int rectype, ret;
 371         u_int32_t new_datalen, old_len;
 372         u_int8_t *hk;
 373
 374         /*
 375          * Check if we can do whatever we need to on this page.  If not,
 376          * then we'll have to move the current element to a new page.
 377          */
 378         hk = H_PAIRDATA(hcp->pagep, hcp->bndx);
 379
 380         /*
 381          * If the item is already off page duplicates or an offpage item,
 382          * then we know we can do whatever we need to do in-place
 383          */
 384         if (HPAGE_PTYPE(hk) == H_OFFDUP || HPAGE_PTYPE(hk) == H_OFFPAGE)
 385                 return (0);
 386
 387         old_len =
 388             LEN_HITEM(hcp->pagep, hashp->hdr->pagesize, H_DATAINDEX(hcp->bndx));
 389         new_datalen = old_len - HKEYDATA_SIZE(0) + add_len;
 390
 391         /*
 392          * We need to add a new page under two conditions:
 393          * 1. The addition makes the total data length cross the BIG
 394          *    threshold and the OFFDUP structure won't fit on this page.
 395          * 2. The addition does not make the total data cross the
 396          *    threshold, but the new data won't fit on the page.
 397          * If neither of these is true, then we can return.
 398          */
 399         if (ISBIG(hashp, new_datalen) && (old_len > HOFFDUP_SIZE ||
 400             HOFFDUP_SIZE - old_len <= P_FREESPACE(hcp->pagep)))
 401                 return (0);
 402
 403         if (!ISBIG(hashp, new_datalen) &&
 404             add_len <= (int32_t)P_FREESPACE(hcp->pagep))
 405                 return (0);
 406
 407         /*
 408          * If we get here, then we need to move the item to a new page.
 409          * Check if there are more pages in the chain.
 410          */
 411
 412         new_datalen = ISBIG(hashp, new_datalen) ?
 413             HOFFDUP_SIZE : HKEYDATA_SIZE(new_datalen);
 414
 415         next_pagep = NULL;
 416         for (next_pgno = NEXT_PGNO(hcp->pagep); next_pgno != PGNO_INVALID;
 417             next_pgno = NEXT_PGNO(next_pagep)) {
 418                 if (next_pagep != NULL &&
 419                     (ret = __ham_put_page(hashp->dbp, next_pagep, 0)) != 0)
 420                         return (ret);
 421
 422                 if ((ret = __ham_get_page(hashp->dbp, next_pgno, &next_pagep)) != 0)
 423                         return (ret);
 424
 425                 if (P_FREESPACE(next_pagep) >= new_datalen)
 426                         break;
 427         }
 428
 429         /* No more pages, add one. */
 430         if (next_pagep == NULL &&
 431             (ret = __ham_add_ovflpage(hashp, hcp->pagep, 0, &next_pagep)) != 0)
 432                 return (ret);
 433
 434         /* Add new page at the end of the chain. */
 435         if (P_FREESPACE(next_pagep) < new_datalen &&
 436             (ret = __ham_add_ovflpage(hashp, next_pagep, 1, &next_pagep)) != 0)
 437                 return (ret);
 438
 439         /* Copy the item to the new page. */
 440         if (DB_LOGGING(hashp->dbp)) {
 441                 rectype = PUTPAIR;
 442                 k.flags = 0;
 443                 d.flags = 0;
 444                 if (HPAGE_PTYPE(
 445                     H_PAIRKEY(hcp->pagep, hcp->bndx)) == H_OFFPAGE) {
 446                         rectype |= PAIR_KEYMASK;
 447                         k.data = H_PAIRKEY(hcp->pagep, hcp->bndx);
 448                         k.size = HOFFPAGE_SIZE;
 449                 } else {
 450                         k.data =
 451                             HKEYDATA_DATA(H_PAIRKEY(hcp->pagep, hcp->bndx));
 452                         k.size = LEN_HKEY(hcp->pagep,
 453                             hashp->hdr->pagesize, hcp->bndx);
 454                 }
 455
 456                 if (HPAGE_PTYPE(hk) == H_OFFPAGE) {
 457                         rectype |= PAIR_DATAMASK;
 458                         d.data = H_PAIRDATA(hcp->pagep, hcp->bndx);
 459                         d.size = HOFFPAGE_SIZE;
 460                 } else {
 461                         d.data =
 462                             HKEYDATA_DATA(H_PAIRDATA(hcp->pagep, hcp->bndx));
 463                         d.size = LEN_HDATA(hcp->pagep,
 464                             hashp->hdr->pagesize, hcp->bndx);
 465                 }
 466
 467
 468                 if ((ret = __ham_insdel_log(hashp->dbp->dbenv->lg_info,
 469                     (DB_TXN *)hashp->dbp->txn, &new_lsn, 0, rectype,
 470                     hashp->dbp->log_fileid, PGNO(next_pagep),
 471                     (u_int32_t)H_NUMPAIRS(next_pagep), &LSN(next_pagep),
 472                     &k, &d)) != 0)
 473                         return (ret);
 474
 475                 /* Move lsn onto page. */
 476                 LSN(next_pagep) = new_lsn;      /* Structure assignment. */
 477         }
 478
 479         __ham_copy_item(hashp, hcp->pagep, H_KEYINDEX(hcp->bndx), next_pagep);
 480         __ham_copy_item(hashp, hcp->pagep, H_DATAINDEX(hcp->bndx), next_pagep);
 481
 482         /* Now delete the pair from the current page. */
 483         ret = __ham_del_pair(hashp, hcp, 0);
 484
 485         (void)__ham_put_page(hashp->dbp, hcp->pagep, 1);
 486         hcp->pagep = next_pagep;
 487         hcp->pgno = PGNO(hcp->pagep);
 488         hcp->bndx = H_NUMPAIRS(hcp->pagep) - 1;
 489         F_SET(hcp, H_EXPAND);
 490         return (ret);
 491 }
 492
 493 /*
 494  * Replace an onpage set of duplicates with the OFFDUP structure that
 495  * references the duplicate page.
 496  * XXX This is really just a special case of __onpage_replace; we should
 497  * probably combine them.
 498  * PUBLIC: void __ham_move_offpage __P((HTAB *, PAGE *, u_int32_t, db_pgno_t));
 499  */
 500 void
 501 __ham_move_offpage(hashp, pagep, ndx, pgno)
 502         HTAB *hashp;
 503         PAGE *pagep;
 504         u_int32_t ndx;
 505         db_pgno_t pgno;
 506 {
 507         DBT new_dbt;
 508         DBT old_dbt;
 509         HOFFDUP od;
 510         db_indx_t i;
 511         int32_t shrink;
 512         u_int8_t *src;
 513
 514         od.type = H_OFFDUP;
 515         od.pgno = pgno;
 516
 517         if (DB_LOGGING(hashp->dbp)) {
 518                 new_dbt.data = &od;
 519                 new_dbt.size = HOFFDUP_SIZE;
 520                 old_dbt.data = P_ENTRY(pagep, ndx);
 521                 old_dbt.size = LEN_HITEM(pagep, hashp->hdr->pagesize, ndx);
 522                 (void)__ham_replace_log(hashp->dbp->dbenv->lg_info,
 523                     (DB_TXN *)hashp->dbp->txn, &LSN(pagep), 0,
 524                     hashp->dbp->log_fileid, PGNO(pagep), (u_int32_t)ndx,
 525                     &LSN(pagep), -1, &old_dbt, &new_dbt, 0);
 526         }
 527
 528         shrink =
 529             LEN_HITEM(pagep, hashp->hdr->pagesize, ndx) - HOFFDUP_SIZE;
 530
 531         if (shrink != 0) {
 532                 /* Copy data. */
 533                 src = (u_int8_t *)(pagep) + HOFFSET(pagep);
 534                 memmove(src + shrink, src, pagep->inp[ndx] - HOFFSET(pagep));
 535                 HOFFSET(pagep) += shrink;
 536
 537                 /* Update index table. */
 538                 for (i = ndx; i < NUM_ENT(pagep); i++)
 539                         pagep->inp[i] += shrink;
 540         }
 541
 542         /* Now copy the offdup entry onto the page. */
 543         memcpy(P_ENTRY(pagep, ndx), &od, HOFFDUP_SIZE);
 544 }