isamc/merge.c

   1 /*
   2  * Copyright (c) 1996-1998, Index Data.
   3  * See the file LICENSE for details.
   4  * Sebastian Hammer, Adam Dickmeiss, Heikki Levanto
   5  *
   6  */
   7
   8 #include <stdlib.h>
   9 #include <assert.h>
  10 #include <string.h>
  11 #include <stdio.h>
  12 #include <log.h>
  13 #include "isamc-p.h"
  14 #include "isamh-p.h"
  15
  16 struct isc_merge_block {
  17     int offset;       /* offset in r_buf */
  18     int block;        /* block number of file (0 if none) */
  19     int dirty;        /* block is different from that on file */
  20 };
  21
  22 static void opt_blocks (ISAMC is, struct isc_merge_block *mb, int ptr,
  23                         int last)
  24 {
  25     int i, no_dirty = 0;
  26     for (i = 0; i<ptr; i++)
  27         if (mb[i].dirty)
  28             no_dirty++;
  29     if (no_dirty*4 < ptr*3)
  30         return;
  31     /* bubble-sort it */
  32     for (i = 0; i<ptr; i++)
  33     {
  34         int tmp, j, j_min = -1;
  35         for (j = i; j<ptr; j++)
  36         {
  37             if (j_min < 0 || mb[j_min].block > mb[j].block)
  38                 j_min = j;
  39         }
  40         assert (j_min >= 0);
  41         tmp = mb[j_min].block;
  42         mb[j_min].block = mb[i].block;
  43         mb[i].block = tmp;
  44         mb[i].dirty = 1;
  45     }
  46     if (!last)
  47         mb[i].dirty = 1;
  48 }
  49
  50 static void flush_blocks (ISAMC is, struct isc_merge_block *mb, int ptr,
  51                           char *r_buf, int *firstpos, int cat, int last,
  52                           int *numkeys)
  53 {
  54     int i;
  55
  56     for (i = 0; i<ptr; i++)
  57     {
  58         /* consider this block number */
  59         if (!mb[i].block)
  60         {
  61             mb[i].block = isc_alloc_block (is, cat);
  62             mb[i].dirty = 1;
  63         }
  64
  65         /* consider next block pointer */
  66         if (last && i == ptr-1)
  67             mb[i+1].block = 0;
  68         else if (!mb[i+1].block)
  69         {
  70             mb[i+1].block = isc_alloc_block (is, cat);
  71             mb[i+1].dirty = 1;
  72             mb[i].dirty = 1;
  73         }
  74     }
  75
  76     for (i = 0; i<ptr; i++)
  77     {
  78         char *src;
  79         ISAMC_BLOCK_SIZE ssize = mb[i+1].offset - mb[i].offset;
  80
  81         assert (ssize);
  82
  83         /* skip rest if not dirty */
  84         if (!mb[i].dirty)
  85         {
  86             assert (mb[i].block);
  87             if (!*firstpos)
  88                 *firstpos = mb[i].block;
  89             if (is->method->debug > 2)
  90                 logf (LOG_LOG, "isc: skip ptr=%d size=%d %d %d",
  91                      i, ssize, cat, mb[i].block);
  92             ++(is->files[cat].no_skip_writes);
  93             continue;
  94         }
  95         /* write block */
  96
  97         if (!*firstpos)
  98         {
  99             *firstpos = mb[i].block;
 100             src = r_buf + mb[i].offset - ISAMC_BLOCK_OFFSET_1;
 101             ssize += ISAMC_BLOCK_OFFSET_1;
 102
 103             memcpy (src+sizeof(int)+sizeof(ssize), numkeys,
 104                     sizeof(*numkeys));
 105             if (is->method->debug > 2)
 106                 logf (LOG_LOG, "isc: flush ptr=%d numk=%d size=%d nextpos=%d",
 107                      i, *numkeys, (int) ssize, mb[i+1].block);
 108         }
 109         else
 110         {
 111             src = r_buf + mb[i].offset - ISAMC_BLOCK_OFFSET_N;
 112             ssize += ISAMC_BLOCK_OFFSET_N;
 113             if (is->method->debug > 2)
 114                 logf (LOG_LOG, "isc: flush ptr=%d size=%d nextpos=%d",
 115                      i, (int) ssize, mb[i+1].block);
 116         }
 117         memcpy (src, &mb[i+1].block, sizeof(int));
 118         memcpy (src+sizeof(int), &ssize, sizeof(ssize));
 119         isc_write_block (is, cat, mb[i].block, src);
 120     }
 121 }
 122
 123 static int get_border (ISAMC is, struct isc_merge_block *mb, int ptr,
 124                        int cat, int firstpos)
 125 {
 126    /* Border set to initial fill or block size depending on
 127       whether we are creating a new one or updating and old one.
 128     */
 129
 130     int fill = mb[ptr].block ? is->method->filecat[cat].bsize :
 131                                is->method->filecat[cat].ifill;
 132     int off = (ptr||firstpos) ? ISAMC_BLOCK_OFFSET_N : ISAMC_BLOCK_OFFSET_1;
 133
 134     assert (ptr < 199);
 135
 136     return mb[ptr].offset + fill - off;
 137 }
 138
 139 ISAMC_P isc_merge (ISAMC is, ISAMC_P ipos, ISAMC_I data)
 140 {
 141
 142     char i_item[128], *i_item_ptr;
 143     int i_more, i_mode, i;
 144
 145     ISAMC_PP pp;
 146     char f_item[128], *f_item_ptr;
 147     int f_more;
 148     int last_dirty = 0;
 149     int debug = is->method->debug;
 150
 151     struct isc_merge_block mb[200];
 152
 153     int firstpos = 0;
 154     int cat = 0;
 155     char r_item_buf[128]; /* temporary result output */
 156     char *r_buf;          /* block with resulting data */
 157     int r_offset = 0;     /* current offset in r_buf */
 158     int ptr = 0;          /* pointer */
 159     void *r_clientData;   /* encode client data */
 160     int border;
 161     int numKeys = 0;
 162
 163     r_clientData = (*is->method->code_start)(ISAMC_ENCODE);
 164     r_buf = is->merge_buf + 128;
 165
 166     pp = isc_pp_open (is, ipos);
 167     /* read first item from file. make sure f_more indicates no boundary */
 168     f_item_ptr = f_item;
 169     f_more = isc_read_item (pp, &f_item_ptr);
 170     if (f_more > 0)
 171         f_more = 1;
 172     cat = pp->cat;
 173
 174     if (debug > 1)
 175         logf (LOG_LOG, "isc: isc_merge begin %d %d", cat, pp->pos);
 176
 177     /* read first item from i */
 178     i_item_ptr = i_item;
 179     i_more = (*data->read_item)(data->clientData, &i_item_ptr, &i_mode);
 180
 181     mb[ptr].block = pp->pos;     /* is zero if no block on disk */
 182     mb[ptr].dirty = 0;
 183     mb[ptr].offset = 0;
 184
 185     border = get_border (is, mb, ptr, cat, firstpos);
 186     while (i_more || f_more)
 187     {
 188         char *r_item = r_item_buf;
 189         int cmp;
 190
 191         if (f_more > 1)
 192         {
 193             /* block to block boundary in the original file. */
 194             f_more = 1;
 195             if (cat == pp->cat)
 196             {
 197                 /* the resulting output is of the same category as the
 198                    the original
 199                 */
 200                 if (r_offset <= mb[ptr].offset +is->method->filecat[cat].mfill)
 201                 {
 202                     /* the resulting output block is too small/empty. Delete
 203                        the original (if any)
 204                     */
 205                     if (debug > 3)
 206                         logf (LOG_LOG, "isc: release A");
 207                     if (mb[ptr].block)
 208                         isc_release_block (is, pp->cat, mb[ptr].block);
 209                     mb[ptr].block = pp->pos;
 210                     if (!mb[ptr].dirty)
 211                         mb[ptr].dirty = 1;
 212                     if (ptr > 0)
 213                         mb[ptr-1].dirty = 1;
 214                 }
 215                 else
 216                 {
 217                     /* indicate new boundary based on the original file */
 218                     mb[++ptr].block = pp->pos;
 219                     mb[ptr].dirty = last_dirty;
 220                     mb[ptr].offset = r_offset;
 221                     if (debug > 3)
 222                         logf (LOG_LOG, "isc: bound ptr=%d,offset=%d",
 223                             ptr, r_offset);
 224                     if (cat==is->max_cat && ptr >= is->method->max_blocks_mem)
 225                     {
 226                         /* We are dealing with block(s) of max size. Block(s)
 227                            except 1 will be flushed.
 228                          */
 229                         if (debug > 2)
 230                             logf (LOG_LOG, "isc: flush A %d sections", ptr);
 231                         flush_blocks (is, mb, ptr-1, r_buf, &firstpos, cat,
 232                                       0, &pp->numKeys);
 233                         mb[0].block = mb[ptr-1].block;
 234                         mb[0].dirty = mb[ptr-1].dirty;
 235                         memcpy (r_buf, r_buf + mb[ptr-1].offset,
 236                                 mb[ptr].offset - mb[ptr-1].offset);
 237                         mb[0].offset = 0;
 238
 239                         mb[1].block = mb[ptr].block;
 240                         mb[1].dirty = mb[ptr].dirty;
 241                         mb[1].offset = mb[ptr].offset - mb[ptr-1].offset;
 242                         ptr = 1;
 243                         r_offset = mb[ptr].offset;
 244                     }
 245                 }
 246             }
 247             border = get_border (is, mb, ptr, cat, firstpos);
 248         }
 249         last_dirty = 0;
 250         if (!f_more)
 251             cmp = -1;
 252         else if (!i_more)
 253             cmp = 1;
 254         else
 255             cmp = (*is->method->compare_item)(i_item, f_item);
 256         if (cmp == 0)                   /* insert i=f */
 257         {
 258             if (!i_mode)   /* delete item? */
 259             {
 260                 /* move i */
 261                 i_item_ptr = i_item;
 262                 i_more = (*data->read_item)(data->clientData, &i_item_ptr,
 263                                            &i_mode);
 264                 /* is next input item the same as current except
 265                    for the delete flag? */
 266                 cmp = (*is->method->compare_item)(i_item, f_item);
 267                 if (!cmp && i_mode)   /* delete/insert nop? */
 268                 {
 269                     /* yes! insert as if it was an insert only */
 270                     memcpy (r_item, i_item, i_item_ptr - i_item);
 271                     i_item_ptr = i_item;
 272                     i_more = (*data->read_item)(data->clientData, &i_item_ptr,
 273                                                 &i_mode);
 274                 }
 275                 else
 276                 {
 277                     /* no! delete the item */
 278                     r_item = NULL;
 279                     last_dirty = 1;
 280                     mb[ptr].dirty = 2;
 281                 }
 282             }
 283             else
 284             {
 285                 memcpy (r_item, f_item, f_item_ptr - f_item);
 286
 287                 /* move i */
 288                 i_item_ptr = i_item;
 289                 i_more = (*data->read_item)(data->clientData, &i_item_ptr,
 290                                            &i_mode);
 291             }
 292             /* move f */
 293             f_item_ptr = f_item;
 294             f_more = isc_read_item (pp, &f_item_ptr);
 295         }
 296         else if (cmp > 0)               /* insert f */
 297         {
 298             memcpy (r_item, f_item, f_item_ptr - f_item);
 299             /* move f */
 300             f_item_ptr = f_item;
 301             f_more = isc_read_item (pp, &f_item_ptr);
 302         }
 303         else                            /* insert i */
 304         {
 305             if (!i_mode)                /* delete item which isn't there? */
 306             {
 307                 logf (LOG_FATAL, "Inconsistent register at offset %d",
 308                                  r_offset);
 309                 abort ();
 310             }
 311             memcpy (r_item, i_item, i_item_ptr - i_item);
 312             mb[ptr].dirty = 2;
 313             last_dirty = 1;
 314             /* move i */
 315             i_item_ptr = i_item;
 316             i_more = (*data->read_item)(data->clientData, &i_item_ptr,
 317                                         &i_mode);
 318         }
 319         if (r_item)  /* insert resulting item? */
 320         {
 321             char *r_out_ptr = r_buf + r_offset;
 322             int new_offset;
 323
 324             (*is->method->code_item)(ISAMC_ENCODE, r_clientData,
 325                                      &r_out_ptr, &r_item);
 326             new_offset = r_out_ptr - r_buf;
 327
 328             numKeys++;
 329
 330             if (border < new_offset && border >= r_offset)
 331             {
 332                 if (debug > 2)
 333                     logf (LOG_LOG, "isc: border %d %d", ptr, border);
 334                 /* Max size of current block category reached ...
 335                    make new virtual block entry */
 336                 mb[++ptr].block = 0;
 337                 mb[ptr].dirty = 1;
 338                 mb[ptr].offset = r_offset;
 339                 if (cat == is->max_cat && ptr >= is->method->max_blocks_mem)
 340                 {
 341                     /* We are dealing with block(s) of max size. Block(s)
 342                        except one will be flushed. Note: the block(s) are
 343                        surely not the last one(s).
 344                      */
 345                     if (debug > 2)
 346                         logf (LOG_LOG, "isc: flush B %d sections", ptr-1);
 347                     flush_blocks (is, mb, ptr-1, r_buf, &firstpos, cat,
 348                                   0, &pp->numKeys);
 349                     mb[0].block = mb[ptr-1].block;
 350                     mb[0].dirty = mb[ptr-1].dirty;
 351                     memcpy (r_buf, r_buf + mb[ptr-1].offset,
 352                             mb[ptr].offset - mb[ptr-1].offset);
 353                     mb[0].offset = 0;
 354
 355                     mb[1].block = mb[ptr].block;
 356                     mb[1].dirty = mb[0].dirty;
 357                     mb[1].offset = mb[ptr].offset - mb[ptr-1].offset;
 358                     memcpy (r_buf + mb[1].offset, r_buf + r_offset,
 359                             new_offset - r_offset);
 360                     new_offset = (new_offset - r_offset) + mb[1].offset;
 361                     ptr = 1;
 362                 }
 363                 border = get_border (is, mb, ptr, cat, firstpos);
 364             }
 365             r_offset = new_offset;
 366         }
 367         if (cat < is->max_cat && ptr >= is->method->filecat[cat].mblocks)
 368         {
 369             /* Max number blocks in current category reached ->
 370                must switch to next category (with larger block size)
 371             */
 372             int j = 0;
 373
 374             (is->files[cat].no_remap)++;
 375             /* delete all original block(s) read so far */
 376             for (i = 0; i < ptr; i++)
 377                 if (mb[i].block)
 378                     isc_release_block (is, pp->cat, mb[i].block);
 379             /* also delete all block to be read in the future */
 380             pp->deleteFlag = 1;
 381
 382             /* remap block offsets */
 383             assert (mb[j].offset == 0);
 384             cat++;
 385             mb[j].dirty = 1;
 386             mb[j].block = 0;
 387             mb[ptr].offset = r_offset;
 388             for (i = 1; i < ptr; i++)
 389             {
 390                 int border = is->method->filecat[cat].ifill -
 391                          ISAMC_BLOCK_OFFSET_1 + mb[j].offset;
 392                 if (debug > 3)
 393                     logf (LOG_LOG, "isc: remap %d border=%d", i, border);
 394                 if (mb[i+1].offset > border && mb[i].offset <= border)
 395                 {
 396                     if (debug > 3)
 397                         logf (LOG_LOG, "isc:  to %d %d", j, mb[i].offset);
 398                     mb[++j].dirty = 1;
 399                     mb[j].block = 0;
 400                     mb[j].offset = mb[i].offset;
 401                 }
 402             }
 403             if (debug > 2)
 404                 logf (LOG_LOG, "isc: remap from %d to %d sections to cat %d",
 405                       ptr, j, cat);
 406             ptr = j;
 407             border = get_border (is, mb, ptr, cat, firstpos);
 408             if (debug > 3)
 409                 logf (LOG_LOG, "isc: border=%d r_offset=%d", border, r_offset);
 410         }
 411     }
 412     if (mb[ptr].offset < r_offset)
 413     {   /* make the final boundary offset */
 414         mb[++ptr].dirty = 1;
 415         mb[ptr].block = 0;
 416         mb[ptr].offset = r_offset;
 417     }
 418     else
 419     {   /* empty output. Release last block if any */
 420         if (cat == pp->cat && mb[ptr].block)
 421         {
 422             if (debug > 3)
 423                 logf (LOG_LOG, "isc: release C");
 424             isc_release_block (is, pp->cat, mb[ptr].block);
 425             mb[ptr].block = 0;
 426             if (ptr > 0)
 427                 mb[ptr-1].dirty = 1;
 428         }
 429     }
 430
 431     if (debug > 2)
 432         logf (LOG_LOG, "isc: flush C, %d sections", ptr);
 433
 434     if (firstpos)
 435     {
 436         /* we have to patch initial block with num keys if that
 437            has changed */
 438         if (numKeys != isc_pp_num (pp))
 439         {
 440             if (debug > 2)
 441                 logf (LOG_LOG, "isc: patch num keys firstpos=%d num=%d",
 442                                 firstpos, numKeys);
 443             bf_write (is->files[cat].bf, firstpos, ISAMC_BLOCK_OFFSET_N,
 444                       sizeof(numKeys), &numKeys);
 445         }
 446     }
 447     else if (ptr > 0)
 448     {   /* we haven't flushed initial block yet and there surely are some
 449            blocks to flush. Make first block dirty if numKeys differ */
 450         if (numKeys != isc_pp_num (pp))
 451             mb[0].dirty = 1;
 452     }
 453     /* flush rest of block(s) in r_buf */
 454     flush_blocks (is, mb, ptr, r_buf, &firstpos, cat, 1, &numKeys);
 455
 456     (*is->method->code_stop)(ISAMC_ENCODE, r_clientData);
 457     if (!firstpos)
 458         cat = 0;
 459     if (debug > 1)
 460         logf (LOG_LOG, "isc: isc_merge return %d %d", cat, firstpos);
 461     isc_pp_close (pp);
 462     return cat + firstpos * 8;
 463 }
 464
 465 static char *hexdump(unsigned char *p, int len, char *buff) {
 466   static char localbuff[128];
 467   char bytebuff[8];
 468   if (!buff) buff=localbuff;
 469   *buff='\0';
 470   while (len--) {
 471     sprintf(bytebuff,"%02x",*p);
 472     p++;
 473     strcat(buff,bytebuff);
 474     if (len) strcat(buff,",");
 475   }
 476   return buff;
 477 }
 478
 479
 480 /* isamh - heikki's append-only isam
 481  * missing:
 482  * read the code from an existing block, to synch encoding
 483  * append to single-block entries
 484  * append to multi-block entries
 485 */
 486
 487 ISAMC_P isamh_append (ISAMH is, ISAMH_P ipos, ISAMH_I data)
 488 {
 489
 490     ISAMH_PP pp;
 491
 492     char i_item[128];
 493     char *i_item_ptr;
 494     int i_more=1, i_mode;
 495
 496     char codebuffer[128];
 497     char *codeptr;
 498     char *bufptr;
 499     int codelen;
 500
 501     ISAMH_PP firstpp;
 502     void *r_clientData;   /* encode client data */
 503     int newblock;
 504     int newcat;
 505     int maxkeys;
 506     int maxsize;
 507     int retval;
 508
 509     pp = firstpp = isamh_pp_open (is, ipos);
 510     assert (*is->method->code_reset);
 511
 512     if ( 0==ipos)
 513     {   /* new block */
 514       pp->cat=0; /* start small... */
 515       pp->pos = isamh_alloc_block(is,pp->cat);
 516       pp->size= pp->offset = ISAMH_BLOCK_OFFSET_1 ;
 517       r_clientData = (*is->method->code_start)(ISAMH_ENCODE);
 518       logf(LOG_LOG,"isamh_append: starting with new block %d",pp->pos);
 519     }
 520     else
 521     { /* existing block */
 522       if (isamh_block(firstpp->lastblock) == firstpp->pos)
 523       { /* only one block, we have it already */
 524         pp->offset=ISAMH_BLOCK_OFFSET_1;
 525         logf(LOG_LOG,"isamh_append: starting with one block %d",pp->pos);
 526       }
 527       else
 528       {
 529         logf(LOG_LOG,"isamh_append: starting with multiple blocks %d>%d>%d",
 530           firstpp->pos,isamh_block(firstpp->next),isamh_block(firstpp->lastblock));
 531         pp=isamh_pp_open(is,firstpp->lastblock);
 532           /* dirty, but this can also read a N-block. Just clear extra values*/
 533         pp->lastblock=0;
 534         pp->offset=ISAMH_BLOCK_OFFSET_N;
 535       } /* get last */
 536       /* read pointers in it to synchronize the encoder ??!! */
 537       r_clientData = (*is->method->code_start)(ISAMH_ENCODE);
 538       logf(LOG_LOG,"isamh_append: scanning to end of block %d %d->%d",
 539         pp->pos, pp->offset, pp->size);
 540       codeptr=codebuffer;
 541       while (pp->offset<pp->size) {
 542         codeptr=codebuffer;
 543         bufptr=pp->buf + pp->offset;
 544         (*is->method->code_item)(ISAMH_DECODE, r_clientData, &codeptr, &bufptr);
 545         codelen = bufptr - (pp->buf+pp->offset) ;
 546         logf(LOG_LOG,"isamh_append: dec at %d %d/%d:%s",
 547           pp->offset, codelen, codeptr-codebuffer,
 548           hexdump(codebuffer,codeptr-codebuffer,0) );
 549         pp->offset += codelen;
 550       }
 551     } /* existing block */
 552
 553
 554     i_item_ptr = i_item;
 555     i_more = (*data->read_item)(data->clientData,&i_item_ptr,&i_mode);
 556     logf(LOG_LOG,"isamh_append 1: m=%d l=%d %s",
 557       i_mode, i_item_ptr-i_item, hexdump(i_item,i_item_ptr-i_item,0));
 558
 559     maxsize = is->method->filecat[pp->cat].bsize;
 560
 561     while(i_more) {
 562        if (i_mode)
 563        { /* insert key, ignore all delete keys time being... */
 564           codeptr = codebuffer;
 565           i_item_ptr=i_item;
 566           (*is->method->code_item)(ISAMH_ENCODE, r_clientData, &codeptr, &i_item_ptr);
 567           codelen = codeptr-codebuffer;
 568
 569           assert( (codelen < 128) && (codelen>0));
 570
 571           logf(LOG_LOG,"isamh_append: coded into %d:%s",
 572                codelen,hexdump(codebuffer,codelen,0));
 573
 574           if ( pp->offset + codelen > maxsize )
 575           {
 576             newcat = pp->cat;
 577             maxkeys = is->method->filecat[pp->cat].mblocks;  /* max keys */
 578             logf(LOG_LOG,"isamh_append: need new block: %d > %d (k:%d/%d)",
 579                 pp->offset + codelen, maxsize, firstpp->numKeys,maxkeys );
 580             if ( (maxkeys>0) && (firstpp->numKeys > maxkeys) )
 581             { /* time to increase block size */
 582                newcat++;
 583                maxsize = is->method->filecat[newcat].bsize;
 584                pp->buf=xrealloc(pp->buf,maxsize);
 585                logf(LOG_LOG,"isamh_append: increased to cat %d ",newcat);
 586             }
 587
 588             newblock = isamh_alloc_block(is,newcat);
 589             pp->next = isamh_addr(newblock,newcat);
 590             if (firstpp!=pp)
 591             {  /* not first block, write to disk already now */
 592               isamh_buildlaterblock(pp);
 593               isamh_write_block(is,pp->cat,pp->pos,pp->buf);
 594               //if (cat != newcat)
 595               //  realloc buf !!!!
 596             }
 597             else
 598             {  /* we had only one block, allocate a second buffer */
 599               pp = isamh_pp_open(is,0);
 600             }
 601             pp->cat = newcat;
 602             pp->pos = newblock;
 603
 604             pp->size=pp->offset=ISAMH_BLOCK_OFFSET_N ;
 605             pp->next=0;
 606             pp->lastblock=0;
 607             logf(LOG_LOG,"isamh_append: got a new block %d:%d",pp->cat,pp->pos);
 608
 609             /* reset the encoding, and code again */
 610             (*is->method->code_reset)(r_clientData);
 611             codeptr = codebuffer;
 612             i_item_ptr=i_item;
 613             (*is->method->code_item)(ISAMH_ENCODE, r_clientData, &codeptr, &i_item_ptr);
 614             codelen = codeptr-codebuffer;
 615             logf(LOG_LOG,"isamh_append: coded again %d:%s",
 616                  codelen,hexdump(codebuffer,codelen,0));
 617
 618           } /* new block needed */
 619
 620           /* ok, now we can write it */
 621           memcpy(&(pp->buf[pp->offset]), codebuffer, codelen);
 622           pp->offset += codelen;
 623           pp->size += codelen;
 624           firstpp->numKeys++;
 625        } /* not a delete */
 626
 627        /* try to read the next element */
 628        i_item_ptr = i_item;
 629        i_more = (*data->read_item)(data->clientData,&i_item_ptr,&i_mode);
 630        logf(LOG_LOG,"isamh_append 2: m=%d l=%d %s",
 631              i_mode, i_item_ptr-i_item, hexdump(i_item,i_item_ptr-i_item,0));
 632
 633     } /* while */
 634
 635     /* Write the last (partial) block, if needed. */
 636     if (pp!=firstpp)
 637     {
 638       isamh_buildlaterblock(pp);
 639       isamh_write_block(is,pp->cat,pp->pos,pp->buf);
 640     }
 641
 642     /* update first block and write it */
 643     firstpp->lastblock = isamh_addr(pp->pos,pp->cat);
 644     isamh_buildfirstblock(firstpp);
 645     isamh_write_block(is,firstpp->cat,firstpp->pos,firstpp->buf);
 646
 647     /* release the second block, if we allocated one */
 648     if ( firstpp != pp )
 649       isamh_pp_close(pp);
 650
 651     /* get return value (before it disappears at close! */
 652     retval = isamh_addr(firstpp->pos,firstpp->cat);
 653
 654     isamh_pp_close(firstpp);
 655
 656     return retval;
 657
 658 } /*  isamh_append */
 659
 660
 661 /*
 662  * $Log: merge.c,v $
 663  * Revision 1.14  1999-07-06 16:30:20  heikki
 664  * IsamH startss to work - at least it builds indexes. Can not search yet...
 665  *
 666  * Revision 1.13  1999/07/06 09:37:05  heikki
 667  * Working on isamh - not ready yet.
 668  *
 669  * Revision 1.12  1999/06/30 15:03:55  heikki
 670  * first take on isamh, the append-only isam structure
 671  *
 672  * Revision 1.11  1999/05/26 07:49:14  adam
 673  * C++ compilation.
 674  *
 675  * Revision 1.10  1998/03/19 12:22:09  adam
 676  * Minor change.
 677  *
 678  * Revision 1.9  1998/03/19 10:04:38  adam
 679  * Minor changes.
 680  *
 681  * Revision 1.8  1998/03/18 09:23:55  adam
 682  * Blocks are stored in chunks on free list - up to factor 2 in speed.
 683  * Fixed bug that could occur in block category rearrangemen.
 684  *
 685  * Revision 1.7  1998/03/11 11:18:18  adam
 686  * Changed the isc_merge to take into account the mfill (minimum-fill).
 687  *
 688  * Revision 1.6  1998/03/06 13:54:03  adam
 689  * Fixed two nasty bugs in isc_merge.
 690  *
 691  * Revision 1.5  1997/02/12 20:42:43  adam
 692  * Bug fix: during isc_merge operations, some pages weren't marked dirty
 693  * even though they should be. At this point the merge operation marks
 694  * a page dirty if the previous page changed at all. A better approach is
 695  * to mark it dirty if the last key written changed in previous page.
 696  *
 697  * Revision 1.4  1996/11/08 11:15:31  adam
 698  * Number of keys in chain are stored in first block and the function
 699  * to retrieve this information, isc_pp_num is implemented.
 700  *
 701  * Revision 1.3  1996/11/04 14:08:59  adam
 702  * Optimized free block usage.
 703  *
 704  * Revision 1.2  1996/11/01 13:36:46  adam
 705  * New element, max_blocks_mem, that control how many blocks of max size
 706  * to store in memory during isc_merge.
 707  * Function isc_merge now ignores delete/update of identical keys and
 708  * the proper blocks are then non-dirty and not written in flush_blocks.
 709  *
 710  * Revision 1.1  1996/11/01  08:59:15  adam
 711  * First version of isc_merge that supports update/delete.
 712  *
 713  */
 714
 715