rset/rsmultiandor.c

   1 /* This file is part of the Zebra server.
   2    Copyright (C) 1994-2011 Index Data
   3
   4 Zebra is free software; you can redistribute it and/or modify it under
   5 the terms of the GNU General Public License as published by the Free
   6 Software Foundation; either version 2, or (at your option) any later
   7 version.
   8
   9 Zebra is distributed in the hope that it will be useful, but WITHOUT ANY
  10 WARRANTY; without even the implied warranty of MERCHANTABILITY or
  11 FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
  12 for more details.
  13
  14 You should have received a copy of the GNU General Public License
  15 along with this program; if not, write to the Free Software
  16 Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
  17
  18 */
  19
  20
  21 /**
  22  * \file rsmultiandor.c
  23  * \brief This module implements the rsmulti_or and rsmulti_and result sets
  24  *
  25  * rsmultior is based on a heap, from which we find the next hit.
  26  *
  27  * rsmultiand is based on a simple array of rsets, and a linear
  28  * search to find the record that exists in all of those rsets.
  29  * To speed things up, the array is sorted so that the smallest
  30  * rsets come first, they are most likely to have the hits furthest
  31  * away, and thus forwarding to them makes the most sense.
  32  */
  33
  34
  35 #if HAVE_CONFIG_H
  36 #include <config.h>
  37 #endif
  38 #include <assert.h>
  39 #include <stdio.h>
  40 #include <stdlib.h>
  41 #include <string.h>
  42
  43 #include <idzebra/util.h>
  44 #include <idzebra/isamc.h>
  45 #include <rset.h>
  46
  47 static RSFD r_open_and (RSET ct, int flag);
  48 static RSFD r_open_or (RSET ct, int flag);
  49 static void r_close (RSFD rfd);
  50 static void r_delete (RSET ct);
  51 static int r_read_and (RSFD rfd, void *buf, TERMID *term);
  52 static int r_read_or (RSFD rfd, void *buf, TERMID *term);
  53 static int r_write (RSFD rfd, const void *buf);
  54 static int r_forward_and(RSFD rfd, void *buf, TERMID *term,
  55                      const void *untilbuf);
  56 static int r_forward_or(RSFD rfd, void *buf, TERMID *term,
  57                      const void *untilbuf);
  58 static void r_pos_and(RSFD rfd, double *current, double *total);
  59 static void r_pos_or(RSFD rfd, double *current, double *total);
  60 static void r_get_terms(RSET ct, TERMID *terms, int maxterms, int *curterm);
  61
  62 static const struct rset_control control_or =
  63 {
  64     "multi-or",
  65     r_delete,
  66     r_get_terms,
  67     r_open_or,
  68     r_close,
  69     r_forward_or,
  70     r_pos_or,
  71     r_read_or,
  72     r_write,
  73 };
  74
  75 static const struct rset_control control_and =
  76 {
  77     "multi-and",
  78     r_delete,
  79     r_get_terms,
  80     r_open_and,
  81     r_close,
  82     r_forward_and,
  83     r_pos_and,
  84     r_read_and,
  85     r_write,
  86 };
  87
  88 /* The heap structure:
  89  * The rset contains a list or rsets we are ORing together
  90  * The rfd contains a heap of heap-items, which contain
  91  * a rfd opened to those rsets, and a buffer for one key.
  92  * They also contain a ptr to the rset list in the rset
  93  * itself, for practical reasons.
  94  */
  95
  96 struct heap_item {
  97     RSFD fd;
  98     void *buf;
  99     RSET rset;
 100     TERMID term;
 101 };
 102
 103 struct heap {
 104     int heapnum;
 105     int heapmax;
 106     const struct rset_key_control *kctrl;
 107     struct heap_item **heap; /* ptrs to the rfd */
 108 };
 109 typedef struct heap *HEAP;
 110
 111
 112 struct rset_private {
 113     int dummy;
 114 };
 115
 116
 117 struct rfd_private {
 118     int flag;
 119     struct heap_item *items; /* we alloc and free them here */
 120     HEAP h; /* and move around here */
 121     zint hits; /* returned so far */
 122     int eof; /* seen the end of it */
 123     int tailcount; /* how many items are tailing */
 124     zint segment;
 125     int skip;
 126     char *tailbits;
 127 };
 128
 129 static int log_level = 0;
 130 static int log_level_initialized = 0;
 131
 132
 133 /* Heap functions ***********************/
 134
 135 #if 0
 136 static void heap_dump_item( HEAP h, int i, int level)
 137 {
 138     double cur,tot;
 139     if (i>h->heapnum)
 140         return;
 141     (void)rset_pos(h->heap[i]->rset,h->heap[i]->fd, &cur, &tot);
 142     yaz_log(log_level," %d %*s i=%p buf=%p %0.1f/%0.1f",i, level, "",
 143                     &(h->heap[i]), h->heap[i]->buf, cur,tot );
 144     heap_dump_item(h, 2*i, level+1);
 145     heap_dump_item(h, 2*i+1, level+1);
 146 }
 147 static void heap_dump( HEAP h,char *msg) {
 148     yaz_log(log_level, "heap dump: %s num=%d max=%d",msg, h->heapnum, h->heapmax);
 149     heap_dump_item(h,1,1);
 150 }
 151 #endif
 152
 153 static void heap_swap (HEAP h, int x, int y)
 154 {
 155     struct heap_item *swap;
 156     swap = h->heap[x];
 157     h->heap[x] = h->heap[y];
 158     h->heap[y] = swap;
 159 }
 160
 161 static int heap_cmp(HEAP h, int x, int y)
 162 {
 163     return (*h->kctrl->cmp)(h->heap[x]->buf,h->heap[y]->buf);
 164 }
 165
 166 static int heap_empty(HEAP h)
 167 {
 168     return ( 0==h->heapnum );
 169 }
 170
 171 /** \brief deletes the first item in the heap, and balances the rest
 172  */
 173 static void heap_delete (HEAP h)
 174 {
 175     int cur = 1, child = 2;
 176     h->heap[1] = 0; /* been deleted */
 177     heap_swap (h, 1, h->heapnum--);
 178     while (child <= h->heapnum) {
 179         if (child < h->heapnum && heap_cmp(h,child,1+child)>0 )
 180             child++;
 181         if (heap_cmp(h,cur,child) > 0)
 182         {
 183             heap_swap (h, cur, child);
 184             cur = child;
 185             child = 2*cur;
 186         }
 187         else
 188             break;
 189     }
 190 }
 191
 192 /** \brief puts item into heap.
 193     The heap root element has changed value (to bigger)
 194     Swap downwards until the heap is ordered again
 195 */
 196 static void heap_balance (HEAP h)
 197 {
 198     int cur = 1, child = 2;
 199     while (child <= h->heapnum) {
 200         if (child < h->heapnum && heap_cmp(h,child,1+child)>0 )
 201             child++;
 202         if (heap_cmp(h,cur,child) > 0)
 203         {
 204             heap_swap (h, cur, child);
 205             cur = child;
 206             child = 2*cur;
 207         }
 208         else
 209             break;
 210     }
 211 }
 212
 213
 214 static void heap_insert (HEAP h, struct heap_item *hi)
 215 {
 216     int cur, parent;
 217
 218     cur = ++(h->heapnum);
 219     assert(cur <= h->heapmax);
 220     h->heap[cur] = hi;
 221     parent = cur/2;
 222     while (parent && (heap_cmp(h,parent,cur) > 0))
 223     {
 224         assert(parent>0);
 225         heap_swap (h, cur, parent);
 226         cur = parent;
 227         parent = cur/2;
 228     }
 229 }
 230
 231
 232 static
 233 HEAP heap_create (NMEM nmem, int size, const struct rset_key_control *kctrl)
 234 {
 235     HEAP h = (HEAP) nmem_malloc (nmem, sizeof(*h));
 236
 237     ++size; /* heap array starts at 1 */
 238     h->heapnum = 0;
 239     h->heapmax = size;
 240     h->kctrl = kctrl;
 241     h->heap = (struct heap_item**) nmem_malloc(nmem,size*sizeof(*h->heap));
 242     h->heap[0]=0; /* not used */
 243     return h;
 244 }
 245
 246 static void heap_clear( HEAP h)
 247 {
 248     assert(h);
 249     h->heapnum = 0;
 250 }
 251
 252 static void heap_destroy (HEAP h)
 253 {
 254     /* nothing to delete, all is nmem'd, and will go away in due time */
 255 }
 256
 257 /** \brief compare and items for quicksort
 258     used in qsort to get the multi-and args in optimal order
 259     that is, those with fewest occurrences first
 260 */
 261 int compare_ands(const void *x, const void *y)
 262 {   const struct heap_item *hx = x;
 263     const struct heap_item *hy = y;
 264     double cur, totx, toty;
 265     rset_pos(hx->fd, &cur, &totx);
 266     rset_pos(hy->fd, &cur, &toty);
 267     if ( totx > toty +0.5 )
 268         return 1;
 269     if ( totx < toty -0.5 )
 270         return -1;
 271     return 0;  /* return totx - toty, except for overflows and rounding */
 272 }
 273
 274 static RSET rsmulti_andor_create(NMEM nmem,
 275                                  struct rset_key_control *kcontrol,
 276                                  int scope, TERMID termid,
 277                                  int no_rsets, RSET* rsets,
 278                                  const struct rset_control *ctrl)
 279 {
 280     RSET rnew = rset_create_base(ctrl, nmem, kcontrol, scope, termid,
 281                                  no_rsets, rsets);
 282     struct rset_private *info;
 283     if (!log_level_initialized)
 284     {
 285         log_level = yaz_log_module_level("rsmultiandor");
 286         log_level_initialized = 1;
 287     }
 288     yaz_log(log_level, "rsmultiand_andor_create scope=%d", scope);
 289     info = (struct rset_private *) nmem_malloc(rnew->nmem, sizeof(*info));
 290     rnew->priv = info;
 291     return rnew;
 292 }
 293
 294 RSET rset_create_or(NMEM nmem, struct rset_key_control *kcontrol,
 295                     int scope, TERMID termid, int no_rsets, RSET* rsets)
 296 {
 297     return rsmulti_andor_create(nmem, kcontrol, scope, termid,
 298                                 no_rsets, rsets, &control_or);
 299 }
 300
 301 RSET rset_create_and(NMEM nmem, struct rset_key_control *kcontrol,
 302                      int scope, int no_rsets, RSET* rsets)
 303 {
 304     return rsmulti_andor_create(nmem, kcontrol, scope, 0,
 305                                 no_rsets, rsets, &control_and);
 306 }
 307
 308 static void r_delete (RSET ct)
 309 {
 310 }
 311
 312 static RSFD r_open_andor (RSET ct, int flag, int is_and)
 313 {
 314     RSFD rfd;
 315     struct rfd_private *p;
 316     const struct rset_key_control *kctrl = ct->keycontrol;
 317     int i;
 318
 319     if (flag & RSETF_WRITE)
 320     {
 321         yaz_log (YLOG_FATAL, "multiandor set type is read-only");
 322         return NULL;
 323     }
 324     rfd = rfd_create_base(ct);
 325     if (rfd->priv) {
 326         p = (struct rfd_private *)rfd->priv;
 327         if (!is_and)
 328             heap_clear(p->h);
 329         assert(p->items);
 330         /* all other pointers shouls already be allocated, in right sizes! */
 331     }
 332     else
 333     {
 334         p = (struct rfd_private *) nmem_malloc (ct->nmem,sizeof(*p));
 335         rfd->priv = p;
 336         p->h = 0;
 337         p->tailbits = 0;
 338         if (is_and)
 339             p->tailbits = nmem_malloc(ct->nmem, ct->no_children*sizeof(char) );
 340         else
 341             p->h = heap_create( ct->nmem, ct->no_children, kctrl);
 342         p->items = (struct heap_item *)
 343             nmem_malloc(ct->nmem, ct->no_children*sizeof(*p->items));
 344         for (i = 0; i<ct->no_children; i++)
 345         {
 346             p->items[i].rset = ct->children[i];
 347             p->items[i].buf = nmem_malloc(ct->nmem, kctrl->key_size);
 348         }
 349     }
 350     p->flag = flag;
 351     p->hits = 0;
 352     p->eof = 0;
 353     p->tailcount = 0;
 354     if (is_and)
 355     { /* read the array and sort it */
 356         for (i = 0; i<ct->no_children; i++){
 357             p->items[i].fd = rset_open(ct->children[i], RSETF_READ);
 358             if (!rset_read(p->items[i].fd, p->items[i].buf, &p->items[i].term))
 359                 p->eof = 1;
 360             p->tailbits[i] = 0;
 361         }
 362         qsort(p->items, ct->no_children, sizeof(p->items[0]), compare_ands);
 363     }
 364     else
 365     { /* fill the heap for ORing */
 366         for (i = 0; i<ct->no_children; i++){
 367             p->items[i].fd = rset_open(ct->children[i],RSETF_READ);
 368             if ( rset_read(p->items[i].fd, p->items[i].buf, &p->items[i].term))
 369                 heap_insert(p->h, &(p->items[i]));
 370         }
 371     }
 372     return rfd;
 373 }
 374
 375 static RSFD r_open_or (RSET ct, int flag)
 376 {
 377     return r_open_andor(ct, flag, 0);
 378 }
 379
 380 static RSFD r_open_and (RSET ct, int flag)
 381 {
 382     return r_open_andor(ct, flag, 1);
 383 }
 384
 385
 386 static void r_close (RSFD rfd)
 387 {
 388     struct rfd_private *p=(struct rfd_private *)(rfd->priv);
 389     int i;
 390
 391     if (p->h)
 392         heap_destroy (p->h);
 393     for (i = 0; i<rfd->rset->no_children; i++)
 394         if (p->items[i].fd)
 395             rset_close(p->items[i].fd);
 396 }
 397
 398 static int r_forward_or(RSFD rfd, void *buf,
 399                         TERMID *term, const void *untilbuf)
 400 { /* while heap head behind untilbuf, forward it and rebalance heap */
 401     struct rfd_private *p = rfd->priv;
 402     const struct rset_key_control *kctrl = rfd->rset->keycontrol;
 403     if (heap_empty(p->h))
 404         return 0;
 405     while ( (*kctrl->cmp)(p->h->heap[1]->buf,untilbuf) < -rfd->rset->scope )
 406     {
 407         if (rset_forward(p->h->heap[1]->fd,p->h->heap[1]->buf,
 408                          &p->h->heap[1]->term, untilbuf))
 409             heap_balance(p->h);
 410         else
 411         {
 412             heap_delete(p->h);
 413             if (heap_empty(p->h))
 414                 return 0;
 415         }
 416
 417     }
 418     return r_read_or(rfd, buf, term);
 419 }
 420
 421
 422 /** \brief reads one item key from an 'or' set
 423     \param rfd set handle
 424     \param buf resulting item buffer
 425     \param term resulting term
 426     \retval 0 EOF
 427     \retval 1 item could be read
 428 */
 429 static int r_read_or (RSFD rfd, void *buf, TERMID *term)
 430 {
 431     RSET rset = rfd->rset;
 432     struct rfd_private *mrfd = rfd->priv;
 433     const struct rset_key_control *kctrl = rset->keycontrol;
 434     struct heap_item *it;
 435     int rdres;
 436     if (heap_empty(mrfd->h))
 437         return 0;
 438     it = mrfd->h->heap[1];
 439     memcpy(buf, it->buf, kctrl->key_size);
 440     if (term)
 441     {
 442         if (rset->term)
 443             *term = rset->term;
 444         else
 445             *term = it->term;
 446     }
 447     (mrfd->hits)++;
 448     rdres = rset_read(it->fd, it->buf, &it->term);
 449     if ( rdres )
 450         heap_balance(mrfd->h);
 451     else
 452         heap_delete(mrfd->h);
 453     return 1;
 454
 455 }
 456
 457 /** \brief reads one item key from an 'and' set
 458     \param rfd set handle
 459     \param buf resulting item buffer
 460     \param term resulting term
 461     \retval 0 EOF
 462     \retval 1 item could be read
 463
 464     Has to return all hits where each item points to the
 465     same sysno (scope), in order. Keep an extra key (hitkey)
 466     as long as all records do not point to hitkey, forward
 467     them, and update hitkey to be the highest seen so far.
 468     (if any item eof's, mark eof, and return 0 thereafter)
 469     Once a hit has been found, scan all items for the smallest
 470     value. Mark all as being in the tail. Read next from that
 471     item, and if not in the same record, clear its tail bit
 472 */
 473 static int r_read_and (RSFD rfd, void *buf, TERMID *term)
 474 {   struct rfd_private *p = rfd->priv;
 475     RSET ct = rfd->rset;
 476     const struct rset_key_control *kctrl = ct->keycontrol;
 477     int i;
 478
 479     while (1) {
 480         if (p->tailcount)
 481         { /* we are tailing, find lowest tail and return it */
 482             int mintail = -1;
 483             int cmp;
 484
 485             for (i = 0; i<ct->no_children; i++)
 486             {
 487                 if (p->tailbits[i])
 488                 {
 489                     if (mintail >= 0)
 490                         cmp = (*kctrl->cmp)
 491                             (p->items[i].buf, p->items[mintail].buf);
 492                     else
 493                         cmp = -1;
 494                     if (cmp < 0)
 495                         mintail = i;
 496
 497                     if (kctrl->get_segment)
 498                     {   /* segments enabled */
 499                         zint segment =  kctrl->get_segment(p->items[i].buf);
 500                         /* store segment if not stored already */
 501                         if (!p->segment && segment)
 502                             p->segment = segment;
 503
 504                         /* skip rest entirely if segments don't match */
 505                         if (p->segment && segment && p->segment != segment)
 506                             p->skip = 1;
 507                     }
 508                 }
 509             }
 510             /* return the lowest tail */
 511             memcpy(buf, p->items[mintail].buf, kctrl->key_size);
 512             if (term)
 513                 *term = p->items[mintail].term;
 514             if (!rset_read(p->items[mintail].fd, p->items[mintail].buf,
 515                            &p->items[mintail].term))
 516             {
 517                 p->eof = 1; /* game over, once tails have been returned */
 518                 p->tailbits[mintail] = 0;
 519                 (p->tailcount)--;
 520             }
 521             else
 522             {
 523                 /* still a tail? */
 524                 cmp = (*kctrl->cmp)(p->items[mintail].buf,buf);
 525                 if (cmp >= rfd->rset->scope)
 526                 {
 527                     p->tailbits[mintail] = 0;
 528                     (p->tailcount)--;
 529                 }
 530             }
 531             if (p->skip)
 532                 continue;  /* skip again.. eventually tailcount will be 0 */
 533             if (p->tailcount == 0)
 534                 (p->hits)++;
 535             return 1;
 536         }
 537         /* not tailing, forward until all records match, and set up */
 538         /* as tails. the earlier 'if' will then return the hits */
 539         if (p->eof)
 540             return 0; /* nothing more to see */
 541         i = 1; /* assume items[0] is highest up */
 542         while (i < ct->no_children)
 543         {
 544             int cmp = (*kctrl->cmp)(p->items[0].buf, p->items[i].buf);
 545             if (cmp <= -rfd->rset->scope) { /* [0] was behind, forward it */
 546                 if (!rset_forward(p->items[0].fd, p->items[0].buf,
 547                                   &p->items[0].term, p->items[i].buf))
 548                 {
 549                     p->eof = 1; /* game over */
 550                     return 0;
 551                 }
 552                 i = 0; /* start forwarding from scratch */
 553             }
 554             else if (cmp>=rfd->rset->scope)
 555             { /* [0] was ahead, forward i */
 556                 if (!rset_forward(p->items[i].fd, p->items[i].buf,
 557                                   &p->items[i].term, p->items[0].buf))
 558                 {
 559                     p->eof = 1; /* game over */
 560                     return 0;
 561                 }
 562             }
 563             else
 564                 i++;
 565         } /* while i */
 566         /* if we get this far, all rsets are now within +- scope of [0] */
 567         /* ergo, we have a hit. Mark them all as tailing, and let the */
 568         /* upper 'if' return the hits in right order */
 569         for (i = 0; i < ct->no_children; i++)
 570             p->tailbits[i] = 1;
 571         p->tailcount = ct->no_children;
 572         p->segment = 0;
 573         p->skip = 0;
 574     } /* while 1 */
 575 }
 576
 577
 578 static int r_forward_and(RSFD rfd, void *buf, TERMID *term,
 579                          const void *untilbuf)
 580 {
 581     struct rfd_private *p = rfd->priv;
 582     RSET ct = rfd->rset;
 583     const struct rset_key_control *kctrl = ct->keycontrol;
 584     int i;
 585     int cmp;
 586     int killtail = 0;
 587
 588     for (i = 0; i<ct->no_children; i++)
 589     {
 590         cmp = (*kctrl->cmp)(p->items[i].buf,untilbuf);
 591         if (cmp <= -rfd->rset->scope)
 592         {
 593             killtail = 1; /* we are moving to a different hit */
 594             if (!rset_forward(p->items[i].fd, p->items[i].buf,
 595                               &p->items[i].term, untilbuf))
 596             {
 597                 p->eof = 1; /* game over */
 598                 p->tailcount = 0;
 599                 return 0;
 600             }
 601         }
 602     }
 603     if (killtail)
 604     {
 605         for (i = 0; i<ct->no_children; i++)
 606             p->tailbits[i] = 0;
 607         p->tailcount = 0;
 608     }
 609     return r_read_and(rfd,buf,term);
 610 }
 611
 612 static void r_pos_x(RSFD rfd, double *current, double *total, int and_op)
 613 {
 614     RSET ct = rfd->rset;
 615     struct rfd_private *mrfd =
 616         (struct rfd_private *)(rfd->priv);
 617     double ratio = and_op ? 0.0 : 1.0;
 618     int i;
 619     double sum_cur = 0.0;
 620     double sum_tot = 0.0;
 621     for (i = 0; i<ct->no_children; i++){
 622         double nratio, cur, tot;
 623         rset_pos(mrfd->items[i].fd, &cur, &tot);
 624         yaz_log(log_level, "r_pos: %d %0.1f %0.1f", i, cur,tot);
 625         nratio = cur / tot;
 626         if (and_op)
 627         {
 628             if (nratio > ratio)
 629                 ratio = nratio;
 630         }
 631         else
 632         {
 633             sum_cur += cur;
 634             sum_tot += tot;
 635         }
 636     }
 637     if (!and_op)
 638         ratio = sum_cur / sum_tot;
 639     if (ratio == 0.0 || ratio == 1.0) { /* nothing there */
 640         *current = 0;
 641         *total = 0;
 642         yaz_log(log_level, "r_pos: NULL  %0.1f %0.1f",  *current, *total);
 643     }
 644     else
 645     {
 646         *current = (double) (mrfd->hits);
 647         *total = *current / ratio;
 648         yaz_log(log_level, "r_pos: =  %0.1f %0.1f",  *current, *total);
 649     }
 650 }
 651
 652 static void r_pos_and(RSFD rfd, double *current, double *total)
 653 {
 654     r_pos_x(rfd, current, total, 1);
 655 }
 656
 657 static void r_pos_or(RSFD rfd, double *current, double *total)
 658 {
 659     r_pos_x(rfd, current, total, 0);
 660 }
 661
 662 static int r_write (RSFD rfd, const void *buf)
 663 {
 664     yaz_log (YLOG_FATAL, "multior set type is read-only");
 665     return -1;
 666 }
 667
 668 static void r_get_terms(RSET ct, TERMID *terms, int maxterms, int *curterm)
 669 {
 670     if (ct->term)
 671         rset_get_one_term(ct, terms, maxterms, curterm);
 672     else
 673     {
 674         /* Special case: Some multi-ors have all terms pointing to the same
 675            term. We do not want to duplicate those. Other multiors (and ands)
 676            have different terms under them. Those we want.
 677         */
 678         int firstterm= *curterm;
 679         int i;
 680
 681         for (i = 0; i<ct->no_children; i++)
 682         {
 683             rset_getterms(ct->children[i], terms, maxterms, curterm);
 684             if ( ( *curterm > firstterm+1 ) &&
 685                  ( *curterm <= maxterms ) &&
 686                  ( terms[(*curterm)-1] == terms[firstterm] )
 687                 )
 688                 (*curterm)--; /* forget the term, seen that before */
 689         }
 690     }
 691 }
 692
 693
 694 /*
 695  * Local variables:
 696  * c-basic-offset: 4
 697  * c-file-style: "Stroustrup"
 698  * indent-tabs-mode: nil
 699  * End:
 700  * vim: shiftwidth=4 tabstop=8 expandtab
 701  */
 702