source4/lib/ldb/ldb_tdb/ldb_index.c

   1 /*
   2    ldb database library
   3
   4    Copyright (C) Andrew Tridgell  2004
   5
   6      ** NOTE! The following LGPL license applies to the ldb
   7      ** library. This does NOT imply that all of Samba is released
   8      ** under the LGPL
   9
  10    This library is free software; you can redistribute it and/or
  11    modify it under the terms of the GNU Lesser General Public
  12    License as published by the Free Software Foundation; either
  13    version 2 of the License, or (at your option) any later version.
  14
  15    This library is distributed in the hope that it will be useful,
  16    but WITHOUT ANY WARRANTY; without even the implied warranty of
  17    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  18    Lesser General Public License for more details.
  19
  20    You should have received a copy of the GNU Lesser General Public
  21    License along with this library; if not, write to the Free Software
  22    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
  23 */
  24
  25 /*
  26  *  Name: ldb
  27  *
  28  *  Component: ldb tdb backend - indexing
  29  *
  30  *  Description: indexing routines for ldb tdb backend
  31  *
  32  *  Author: Andrew Tridgell
  33  */
  34
  35 #include "includes.h"
  36 #include "ldb/ldb_tdb/ldb_tdb.h"
  37 #include "ldb/include/ldb_parse.h"
  38
  39 struct dn_list {
  40         unsigned int count;
  41         char **dn;
  42 };
  43
  44 /*
  45   free a struct dn_list
  46 */
  47 static void dn_list_free(struct ldb_context *ldb, struct dn_list *list)
  48 {
  49         int i;
  50         for (i=0;i<list->count;i++) {
  51                 ldb_free(ldb, list->dn[i]);
  52         }
  53         ldb_free(ldb, list->dn);
  54 }
  55
  56 /*
  57   return the dn key to be used for an index
  58   caller frees
  59 */
  60 static char *ldb_dn_key(struct ldb_context *ldb,
  61                         const char *attr, const struct ldb_val *value)
  62 {
  63         char *ret = NULL;
  64
  65         if (ldb_should_b64_encode(value)) {
  66                 char *vstr = ldb_base64_encode(ldb, value->data, value->length);
  67                 if (!vstr) return NULL;
  68                 ldb_asprintf(ldb, &ret, "%s:%s::%s", LTDB_INDEX, attr, vstr);
  69                 ldb_free(ldb, vstr);
  70                 return ret;
  71         }
  72
  73         ldb_asprintf(ldb, &ret, "%s:%s:%s", LTDB_INDEX, attr, (char *)value->data);
  74         return ret;
  75 }
  76
  77 /*
  78   see if a attribute value is in the list of indexed attributes
  79 */
  80 static int ldb_msg_find_idx(const struct ldb_message *msg, const char *attr,
  81                             int *v_idx, const char *key)
  82 {
  83         int i, j;
  84         for (i=0;i<msg->num_elements;i++) {
  85                 if (ldb_attr_cmp(msg->elements[i].name, key) == 0) {
  86                         const struct ldb_message_element *el =
  87                                 &msg->elements[i];
  88                         for (j=0;j<el->num_values;j++) {
  89                                 if (ldb_attr_cmp((char *)el->values[j].data, attr) == 0) {
  90                                         if (v_idx) {
  91                                                 *v_idx = j;
  92                                         }
  93                                         return i;
  94                                 }
  95                         }
  96                 }
  97         }
  98         return -1;
  99 }
 100
 101 /* used in sorting dn lists */
 102 static int list_cmp(const char **s1, const char **s2)
 103 {
 104         return strcmp(*s1, *s2);
 105 }
 106
 107 /*
 108   return a list of dn's that might match a simple indexed search or
 109  */
 110 static int ltdb_index_dn_simple(struct ldb_context *ldb,
 111                                 struct ldb_parse_tree *tree,
 112                                 const struct ldb_message *index_list,
 113                                 struct dn_list *list)
 114 {
 115         char *dn = NULL;
 116         int ret, i, j;
 117         struct ldb_message msg;
 118
 119         list->count = 0;
 120         list->dn = NULL;
 121
 122         /*
 123           if the value is a wildcard then we can't do a match via indexing
 124         */
 125         if (ltdb_has_wildcard(ldb, tree->u.simple.attr, &tree->u.simple.value)) {
 126                 return -1;
 127         }
 128
 129         /* if the attribute isn't in the list of indexed attributes then
 130            this node needs a full search */
 131         if (ldb_msg_find_idx(index_list, tree->u.simple.attr, NULL, LTDB_IDXATTR) == -1) {
 132                 return -1;
 133         }
 134
 135         /* the attribute is indexed. Pull the list of DNs that match the
 136            search criterion */
 137         dn = ldb_dn_key(ldb, tree->u.simple.attr, &tree->u.simple.value);
 138         if (!dn) return -1;
 139
 140         ret = ltdb_search_dn1(ldb, dn, &msg);
 141         ldb_free(ldb, dn);
 142         if (ret == 0 || ret == -1) {
 143                 return ret;
 144         }
 145
 146         for (i=0;i<msg.num_elements;i++) {
 147                 struct ldb_message_element *el;
 148
 149                 if (strcmp(msg.elements[i].name, LTDB_IDX) != 0) {
 150                         continue;
 151                 }
 152
 153                 el = &msg.elements[i];
 154
 155                 list->dn = ldb_malloc_array_p(ldb, char *, el->num_values);
 156                 if (!list->dn) {
 157                         break;
 158                 }
 159
 160                 for (j=0;j<el->num_values;j++) {
 161                         list->dn[list->count] =
 162                                 ldb_strdup(ldb, (char *)el->values[j].data);
 163                         if (!list->dn[list->count]) {
 164                                 dn_list_free(ldb, list);
 165                                 ltdb_search_dn1_free(ldb, &msg);
 166                                 return -1;
 167                         }
 168                         list->count++;
 169                 }
 170         }
 171
 172         ltdb_search_dn1_free(ldb, &msg);
 173
 174         qsort(list->dn, list->count, sizeof(char *), (comparison_fn_t) list_cmp);
 175
 176         return 1;
 177 }
 178
 179 /*
 180   return a list of dn's that might match a simple indexed search on
 181   the special objectclass attribute
 182  */
 183 static int ltdb_index_dn_objectclass(struct ldb_context *ldb,
 184                                      struct ldb_parse_tree *tree,
 185                                      const struct ldb_message *index_list,
 186                                      struct dn_list *list)
 187 {
 188         struct ltdb_private *ltdb = ldb->private_data;
 189         int i;
 190         int ret;
 191         const char *target = tree->u.simple.value.data;
 192         static int list_union(struct ldb_context *,
 193                               struct dn_list *, const struct dn_list *);
 194
 195         list->count = 0;
 196         list->dn = NULL;
 197
 198         ret = ltdb_index_dn_simple(ldb, tree, index_list, list);
 199
 200         for (i=0;i<ltdb->cache.subclasses.num_elements;i++) {
 201                 struct ldb_message_element *el = &ltdb->cache.subclasses.elements[i];
 202                 if (ldb_attr_cmp(el->name, target) == 0) {
 203                         int j;
 204                         for (j=0;j<el->num_values;j++) {
 205                                 struct ldb_parse_tree tree2;
 206                                 struct dn_list list2;
 207                                 tree2.operation = LDB_OP_SIMPLE;
 208                                 tree2.u.simple.attr = ldb_strdup(ldb, LTDB_OBJECTCLASS);
 209                                 if (!tree2.u.simple.attr) {
 210                                         return -1;
 211                                 }
 212                                 tree2.u.simple.value = el->values[j];
 213                                 if (ltdb_index_dn_objectclass(ldb, &tree2,
 214                                                               index_list, &list2) == 1) {
 215                                         if (list->count == 0) {
 216                                                 *list = list2;
 217                                                 ret = 1;
 218                                         } else {
 219                                                 list_union(ldb, list, &list2);
 220                                                 dn_list_free(ldb, &list2);
 221                                         }
 222                                 }
 223                                 ldb_free(ldb, tree2.u.simple.attr);
 224                         }
 225                 }
 226         }
 227
 228         return ret;
 229 }
 230
 231 /*
 232   return a list of dn's that might match a leaf indexed search
 233  */
 234 static int ltdb_index_dn_leaf(struct ldb_context *ldb,
 235                               struct ldb_parse_tree *tree,
 236                               const struct ldb_message *index_list,
 237                               struct dn_list *list)
 238 {
 239         if (ldb_attr_cmp(tree->u.simple.attr, LTDB_OBJECTCLASS) == 0) {
 240                 return ltdb_index_dn_objectclass(ldb, tree, index_list, list);
 241         }
 242         return ltdb_index_dn_simple(ldb, tree, index_list, list);
 243 }
 244
 245
 246 /*
 247   list intersection
 248   list = list & list2
 249   relies on the lists being sorted
 250 */
 251 static int list_intersect(struct ldb_context *ldb,
 252                           struct dn_list *list, const struct dn_list *list2)
 253 {
 254         struct dn_list list3;
 255         int i;
 256
 257         if (list->count == 0 || list2->count == 0) {
 258                 /* 0 & X == 0 */
 259                 dn_list_free(ldb, list);
 260                 return 0;
 261         }
 262
 263         list3.dn = ldb_malloc_array_p(ldb, char *, list->count);
 264         if (!list3.dn) {
 265                 dn_list_free(ldb, list);
 266                 return -1;
 267         }
 268         list3.count = 0;
 269
 270         for (i=0;i<list->count;i++) {
 271                 if (list_find(list->dn[i], list2->dn, list2->count,
 272                               sizeof(char *), (comparison_fn_t)strcmp) != -1) {
 273                         list3.dn[list3.count] = list->dn[i];
 274                         list3.count++;
 275                 } else {
 276                         ldb_free(ldb, list->dn[i]);
 277                 }
 278         }
 279
 280         ldb_free(ldb, list->dn);
 281         list->dn = list3.dn;
 282         list->count = list3.count;
 283
 284         return 0;
 285 }
 286
 287
 288 /*
 289   list union
 290   list = list | list2
 291   relies on the lists being sorted
 292 */
 293 static int list_union(struct ldb_context *ldb,
 294                       struct dn_list *list, const struct dn_list *list2)
 295 {
 296         int i;
 297         char **d;
 298         unsigned int count = list->count;
 299
 300         if (list->count == 0 && list2->count == 0) {
 301                 /* 0 | 0 == 0 */
 302                 dn_list_free(ldb, list);
 303                 return 0;
 304         }
 305
 306         d = ldb_realloc_p(ldb, list->dn, char *, list->count + list2->count);
 307         if (!d) {
 308                 dn_list_free(ldb, list);
 309                 return -1;
 310         }
 311         list->dn = d;
 312
 313         for (i=0;i<list2->count;i++) {
 314                 if (list_find(list2->dn[i], list->dn, count,
 315                               sizeof(char *), (comparison_fn_t)strcmp) == -1) {
 316                         list->dn[list->count] = ldb_strdup(ldb, list2->dn[i]);
 317                         if (!list->dn[list->count]) {
 318                                 dn_list_free(ldb, list);
 319                                 return -1;
 320                         }
 321                         list->count++;
 322                 }
 323         }
 324
 325         if (list->count != count) {
 326                 qsort(list->dn, list->count, sizeof(char *), (comparison_fn_t)list_cmp);
 327         }
 328
 329         return 0;
 330 }
 331
 332 static int ltdb_index_dn(struct ldb_context *ldb,
 333                          struct ldb_parse_tree *tree,
 334                          const struct ldb_message *index_list,
 335                          struct dn_list *list);
 336
 337
 338 /*
 339   OR two index results
 340  */
 341 static int ltdb_index_dn_or(struct ldb_context *ldb,
 342                             struct ldb_parse_tree *tree,
 343                             const struct ldb_message *index_list,
 344                             struct dn_list *list)
 345 {
 346         int ret, i;
 347
 348         ret = -1;
 349         list->dn = NULL;
 350         list->count = 0;
 351
 352         for (i=0;i<tree->u.list.num_elements;i++) {
 353                 struct dn_list list2;
 354                 int v;
 355                 v = ltdb_index_dn(ldb, tree->u.list.elements[i], index_list, &list2);
 356
 357                 if (v == 0) {
 358                         /* 0 || X == X */
 359                         if (ret == -1) {
 360                                 ret = 0;
 361                         }
 362                         continue;
 363                 }
 364
 365                 if (v == -1) {
 366                         /* 1 || X == 1 */
 367                         dn_list_free(ldb, list);
 368                         return -1;
 369                 }
 370
 371                 if (ret == -1) {
 372                         ret = 1;
 373                         *list = list2;
 374                 } else {
 375                         if (list_union(ldb, list, &list2) == -1) {
 376                                 dn_list_free(ldb, &list2);
 377                                 return -1;
 378                         }
 379                         dn_list_free(ldb, &list2);
 380                 }
 381         }
 382
 383         if (list->count == 0) {
 384                 dn_list_free(ldb, list);
 385                 return 0;
 386         }
 387
 388         return ret;
 389 }
 390
 391
 392 /*
 393   NOT an index results
 394  */
 395 static int ltdb_index_dn_not(struct ldb_context *ldb,
 396                              struct ldb_parse_tree *tree,
 397                              const struct ldb_message *index_list,
 398                              struct dn_list *list)
 399 {
 400         /* the only way to do an indexed not would be if we could
 401            negate the not via another not or if we knew the total
 402            number of database elements so we could know that the
 403            existing expression covered the whole database.
 404
 405            instead, we just give up, and rely on a full index scan
 406            (unless an outer & manages to reduce the list)
 407         */
 408         return -1;
 409 }
 410
 411 /*
 412   AND two index results
 413  */
 414 static int ltdb_index_dn_and(struct ldb_context *ldb,
 415                              struct ldb_parse_tree *tree,
 416                              const struct ldb_message *index_list,
 417                              struct dn_list *list)
 418 {
 419         int ret, i;
 420
 421         ret = -1;
 422         list->dn = NULL;
 423         list->count = 0;
 424
 425         for (i=0;i<tree->u.list.num_elements;i++) {
 426                 struct dn_list list2;
 427                 int v;
 428                 v = ltdb_index_dn(ldb, tree->u.list.elements[i], index_list, &list2);
 429
 430                 if (v == 0) {
 431                         /* 0 && X == 0 */
 432                         dn_list_free(ldb, list);
 433                         return 0;
 434                 }
 435
 436                 if (v == -1) {
 437                         continue;
 438                 }
 439
 440                 if (ret == -1) {
 441                         ret = 1;
 442                         *list = list2;
 443                 } else {
 444                         if (list_intersect(ldb, list, &list2) == -1) {
 445                                 dn_list_free(ldb, &list2);
 446                                 return -1;
 447                         }
 448                         dn_list_free(ldb, &list2);
 449                 }
 450
 451                 if (list->count == 0) {
 452                         if (list->dn) ldb_free(ldb, list->dn);
 453                         return 0;
 454                 }
 455         }
 456
 457         return ret;
 458 }
 459
 460 /*
 461   return a list of dn's that might match a indexed search or
 462   -1 if an error. return 0 for no matches, or 1 for matches
 463  */
 464 static int ltdb_index_dn(struct ldb_context *ldb,
 465                          struct ldb_parse_tree *tree,
 466                          const struct ldb_message *index_list,
 467                          struct dn_list *list)
 468 {
 469         int ret = -1;
 470
 471         switch (tree->operation) {
 472         case LDB_OP_SIMPLE:
 473                 ret = ltdb_index_dn_leaf(ldb, tree, index_list, list);
 474                 break;
 475
 476         case LDB_OP_AND:
 477                 ret = ltdb_index_dn_and(ldb, tree, index_list, list);
 478                 break;
 479
 480         case LDB_OP_OR:
 481                 ret = ltdb_index_dn_or(ldb, tree, index_list, list);
 482                 break;
 483
 484         case LDB_OP_NOT:
 485                 ret = ltdb_index_dn_not(ldb, tree, index_list, list);
 486                 break;
 487         }
 488
 489         return ret;
 490 }
 491
 492 /*
 493   filter a candidate dn_list from an indexed search into a set of results
 494   extracting just the given attributes
 495 */
 496 static int ldb_index_filter(struct ldb_context *ldb, struct ldb_parse_tree *tree,
 497                             const char *base,
 498                             enum ldb_scope scope,
 499                             const struct dn_list *dn_list,
 500                             const char * const attrs[], struct ldb_message ***res)
 501 {
 502         int i;
 503         unsigned int count = 0;
 504
 505         for (i=0;i<dn_list->count;i++) {
 506                 struct ldb_message msg;
 507                 int ret;
 508                 ret = ltdb_search_dn1(ldb, dn_list->dn[i], &msg);
 509                 if (ret == 0) {
 510                         /* the record has disappeared? yes, this can happen */
 511                         continue;
 512                 }
 513
 514                 if (ret == -1) {
 515                         /* an internal error */
 516                         return -1;
 517                 }
 518
 519                 if (ldb_message_match(ldb, &msg, tree, base, scope) == 1) {
 520                         ret = ltdb_add_attr_results(ldb, &msg, attrs, &count, res);
 521                 }
 522                 ltdb_search_dn1_free(ldb, &msg);
 523                 if (ret != 0) {
 524                         return -1;
 525                 }
 526         }
 527
 528         return count;
 529 }
 530
 531 /*
 532   search the database with a LDAP-like expression using indexes
 533   returns -1 if an indexed search is not possible, in which
 534   case the caller should call ltdb_search_full()
 535 */
 536 int ltdb_search_indexed(struct ldb_context *ldb,
 537                         const char *base,
 538                         enum ldb_scope scope,
 539                         struct ldb_parse_tree *tree,
 540                         const char * const attrs[], struct ldb_message ***res)
 541 {
 542         struct ltdb_private *ltdb = ldb->private_data;
 543         struct dn_list dn_list;
 544         int ret;
 545
 546         if (ltdb->cache.indexlist.num_elements == 0) {
 547                 /* no index list? must do full search */
 548                 return -1;
 549         }
 550
 551         ret = ltdb_index_dn(ldb, tree, &ltdb->cache.indexlist, &dn_list);
 552
 553         if (ret == 1) {
 554                 /* we've got a candidate list - now filter by the full tree
 555                    and extract the needed attributes */
 556                 ret = ldb_index_filter(ldb, tree, base, scope, &dn_list,
 557                                        attrs, res);
 558                 dn_list_free(ldb, &dn_list);
 559         }
 560
 561         return ret;
 562 }
 563
 564 /*
 565   add a index element where this is the first indexed DN for this value
 566 */
 567 static int ltdb_index_add1_new(struct ldb_context *ldb,
 568                                struct ldb_message *msg,
 569                                struct ldb_message_element *el,
 570                                char *dn)
 571 {
 572         struct ldb_message_element *el2;
 573
 574         /* add another entry */
 575         el2 = ldb_realloc_p(ldb, msg->elements,
 576                             struct ldb_message_element, msg->num_elements+1);
 577         if (!el2) {
 578                 return -1;
 579         }
 580
 581         msg->elements = el2;
 582         msg->elements[msg->num_elements].name = ldb_strdup(ldb, LTDB_IDX);
 583         if (!msg->elements[msg->num_elements].name) {
 584                 return -1;
 585         }
 586         msg->elements[msg->num_elements].num_values = 0;
 587         msg->elements[msg->num_elements].values = ldb_malloc_p(ldb, struct ldb_val);
 588         if (!msg->elements[msg->num_elements].values) {
 589                 return -1;
 590         }
 591         msg->elements[msg->num_elements].values[0].length = strlen(dn);
 592         msg->elements[msg->num_elements].values[0].data = dn;
 593         msg->elements[msg->num_elements].num_values = 1;
 594         msg->num_elements++;
 595
 596         return 0;
 597 }
 598
 599
 600 /*
 601   add a index element where this is not the first indexed DN for this
 602   value
 603 */
 604 static int ltdb_index_add1_add(struct ldb_context *ldb,
 605                                struct ldb_message *msg,
 606                                struct ldb_message_element *el,
 607                                int idx,
 608                                char *dn)
 609 {
 610         struct ldb_val *v2;
 611         int i;
 612
 613         /* for multi-valued attributes we can end up with repeats */
 614         for (i=0;i<msg->elements[idx].num_values;i++) {
 615                 if (strcmp(dn, msg->elements[idx].values[i].data) == 0) {
 616                         return 0;
 617                 }
 618         }
 619
 620         v2 = ldb_realloc_p(ldb, msg->elements[idx].values,
 621                            struct ldb_val,
 622                            msg->elements[idx].num_values+1);
 623         if (!v2) {
 624                 return -1;
 625         }
 626         msg->elements[idx].values = v2;
 627
 628         msg->elements[idx].values[msg->elements[idx].num_values].length = strlen(dn);
 629         msg->elements[idx].values[msg->elements[idx].num_values].data = dn;
 630         msg->elements[idx].num_values++;
 631
 632         return 0;
 633 }
 634
 635 /*
 636   add an index entry for one message element
 637 */
 638 static int ltdb_index_add1(struct ldb_context *ldb, char *dn,
 639                            struct ldb_message_element *el, int v_idx)
 640 {
 641         struct ldb_message msg;
 642         char *dn_key;
 643         int ret, i, added=0, added_dn=0;
 644
 645         dn_key = ldb_dn_key(ldb, el->name, &el->values[v_idx]);
 646         if (!dn_key) {
 647                 return -1;
 648         }
 649
 650         ret = ltdb_search_dn1(ldb, dn_key, &msg);
 651         if (ret == -1) {
 652                 ldb_free(ldb, dn_key);
 653                 return -1;
 654         }
 655
 656         if (ret == 0) {
 657                 added_dn = 1;
 658                 msg.dn = ldb_strdup(ldb, dn_key);
 659                 if (!msg.dn) {
 660                         ldb_free(ldb, dn_key);
 661                         errno = ENOMEM;
 662                         return -1;
 663                 }
 664                 msg.num_elements = 0;
 665                 msg.elements = NULL;
 666                 msg.private_data = NULL;
 667         }
 668
 669         ldb_free(ldb, dn_key);
 670
 671         for (i=0;i<msg.num_elements;i++) {
 672                 if (strcmp(LTDB_IDX, msg.elements[i].name) == 0) {
 673                         break;
 674                 }
 675         }
 676
 677         if (i == msg.num_elements) {
 678                 added = 1;
 679                 ret = ltdb_index_add1_new(ldb, &msg, el, dn);
 680         } else {
 681                 ret = ltdb_index_add1_add(ldb, &msg, el, i, dn);
 682         }
 683
 684         if (ret == 0) {
 685                 ret = ltdb_store(ldb, &msg, TDB_REPLACE);
 686         }
 687
 688         if (added) {
 689                 ldb_free(ldb, msg.elements[i].name);
 690         }
 691         if (added_dn) {
 692                 ldb_free(ldb, msg.dn);
 693         }
 694
 695         ltdb_search_dn1_free(ldb, &msg);
 696
 697         return ret;
 698 }
 699
 700 /*
 701   add the index entries for a new record
 702   return -1 on failure
 703 */
 704 int ltdb_index_add(struct ldb_context *ldb, const struct ldb_message *msg)
 705 {
 706         struct ltdb_private *ltdb = ldb->private_data;
 707         int ret, i, j;
 708
 709         if (ltdb->cache.indexlist.num_elements == 0) {
 710                 /* no indexed fields */
 711                 return 0;
 712         }
 713
 714         for (i=0;i<msg->num_elements;i++) {
 715                 ret = ldb_msg_find_idx(&ltdb->cache.indexlist, msg->elements[i].name,
 716                                        NULL, LTDB_IDXATTR);
 717                 if (ret == -1) {
 718                         continue;
 719                 }
 720                 for (j=0;j<msg->elements[i].num_values;j++) {
 721                         ret = ltdb_index_add1(ldb, msg->dn, &msg->elements[i], j);
 722                         if (ret == -1) {
 723                                 return -1;
 724                         }
 725                 }
 726         }
 727
 728         return 0;
 729 }
 730
 731
 732 /*
 733   delete an index entry for one message element
 734 */
 735 static int ltdb_index_del1(struct ldb_context *ldb, const char *dn,
 736                            struct ldb_message_element *el, int v_idx)
 737 {
 738         struct ldb_message msg;
 739         char *dn_key;
 740         int ret, i, j;
 741
 742         dn_key = ldb_dn_key(ldb, el->name, &el->values[v_idx]);
 743         if (!dn_key) {
 744                 return -1;
 745         }
 746
 747         ret = ltdb_search_dn1(ldb, dn_key, &msg);
 748         if (ret == -1) {
 749                 ldb_free(ldb, dn_key);
 750                 return -1;
 751         }
 752
 753         if (ret == 0) {
 754                 /* it wasn't indexed. Did we have an earlier error? If we did then
 755                    its gone now */
 756                 ldb_debug(ldb, LDB_DEBUG_ERROR, "ERROR: dn_key %s was not indexed\n", dn_key);
 757                 ldb_free(ldb, dn_key);
 758                 return 0;
 759         }
 760
 761         i = ldb_msg_find_idx(&msg, dn, &j, LTDB_IDX);
 762         if (i == -1) {
 763                 ldb_debug(ldb, LDB_DEBUG_ERROR, "ERROR: dn %s not found in %s\n", dn, dn_key);
 764                 /* it ain't there. hmmm */
 765                 ltdb_search_dn1_free(ldb, &msg);
 766                 ldb_free(ldb, dn_key);
 767                 return 0;
 768         }
 769
 770         if (j != msg.elements[i].num_values - 1) {
 771                 memmove(&msg.elements[i].values[j],
 772                         &msg.elements[i].values[j+1],
 773                         (msg.elements[i].num_values-(j+1)) *
 774                         sizeof(msg.elements[i].values[0]));
 775         }
 776         msg.elements[i].num_values--;
 777
 778         if (msg.elements[i].num_values == 0) {
 779                 ret = ltdb_delete_noindex(ldb, dn_key);
 780         } else {
 781                 ret = ltdb_store(ldb, &msg, TDB_REPLACE);
 782         }
 783
 784         ltdb_search_dn1_free(ldb, &msg);
 785         ldb_free(ldb, dn_key);
 786
 787         return ret;
 788 }
 789
 790 /*
 791   delete the index entries for a record
 792   return -1 on failure
 793 */
 794 int ltdb_index_del(struct ldb_context *ldb, const struct ldb_message *msg)
 795 {
 796         struct ltdb_private *ltdb = ldb->private_data;
 797         int ret, i, j;
 798
 799         /* find the list of indexed fields */
 800         if (ltdb->cache.indexlist.num_elements == 0) {
 801                 /* no indexed fields */
 802                 return 0;
 803         }
 804
 805         for (i=0;i<msg->num_elements;i++) {
 806                 ret = ldb_msg_find_idx(&ltdb->cache.indexlist, msg->elements[i].name,
 807                                        NULL, LTDB_IDXATTR);
 808                 if (ret == -1) {
 809                         continue;
 810                 }
 811                 for (j=0;j<msg->elements[i].num_values;j++) {
 812                         ret = ltdb_index_del1(ldb, msg->dn, &msg->elements[i], j);
 813                         if (ret == -1) {
 814                                 return -1;
 815                         }
 816                 }
 817         }
 818
 819         return 0;
 820 }
 821
 822
 823 /*
 824   traversal function that deletes all @INDEX records
 825 */
 826 static int delete_index(struct tdb_context *tdb, TDB_DATA key, TDB_DATA data, void *state)
 827 {
 828         const char *dn = "DN=" LTDB_INDEX ":";
 829         if (strncmp(key.dptr, dn, strlen(dn)) == 0) {
 830                 return tdb_delete(tdb, key);
 831         }
 832         return 0;
 833 }
 834
 835 /*
 836   traversal function that adds @INDEX records during a re index
 837 */
 838 static int re_index(struct tdb_context *tdb, TDB_DATA key, TDB_DATA data, void *state)
 839 {
 840         struct ldb_context *ldb = state;
 841         struct ldb_message msg;
 842         int ret;
 843
 844         if (strncmp(key.dptr, "DN=@", 4) == 0 ||
 845             strncmp(key.dptr, "DN=", 3) != 0) {
 846                 return 0;
 847         }
 848
 849         ret = ltdb_unpack_data(ldb, &data, &msg);
 850         if (ret != 0) {
 851                 return -1;
 852         }
 853
 854         if (!msg.dn) {
 855                 msg.dn = key.dptr+3;
 856         }
 857
 858         ret = ltdb_index_add(ldb, &msg);
 859
 860         ltdb_unpack_data_free(ldb, &msg);
 861
 862         return ret;
 863 }
 864
 865 /*
 866   force a complete reindex of the database
 867 */
 868 int ltdb_reindex(struct ldb_context *ldb)
 869 {
 870         struct ltdb_private *ltdb = ldb->private_data;
 871         int ret;
 872
 873         ltdb_cache_free(ldb);
 874
 875         if (ltdb_cache_load(ldb) != 0) {
 876                 return -1;
 877         }
 878
 879         /* first traverse the database deleting any @INDEX records */
 880         ret = tdb_traverse(ltdb->tdb, delete_index, NULL);
 881         if (ret == -1) {
 882                 errno = EIO;
 883                 return -1;
 884         }
 885
 886         /* now traverse adding any indexes for normal LDB records */
 887         ret = tdb_traverse(ltdb->tdb, re_index, ldb);
 888         if (ret == -1) {
 889                 errno = EIO;
 890                 return -1;
 891         }
 892
 893         return 0;
 894 }