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