rset/rsmultiandor.c

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