source4/lib/ldb/ldb_tdb/ldb_index.c

   1 /*
   2    ldb database library
   3
   4    Copyright (C) Andrew Tridgell  2004-2009
   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 3 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, see <http://www.gnu.org/licenses/>.
  22 */
  23
  24 /*
  25  *  Name: ldb
  26  *
  27  *  Component: ldb tdb backend - indexing
  28  *
  29  *  Description: indexing routines for ldb tdb backend
  30  *
  31  *  Author: Andrew Tridgell
  32  */
  33
  34 #include "ldb_tdb.h"
  35
  36 struct dn_list {
  37         unsigned int count;
  38         struct ldb_val *dn;
  39 };
  40
  41 struct ltdb_idxptr {
  42         struct tdb_context *itdb;
  43         int error;
  44 };
  45
  46 /* we put a @IDXVERSION attribute on index entries. This
  47    allows us to tell if it was written by an older version
  48 */
  49 #define LTDB_INDEXING_VERSION 2
  50
  51 /* enable the idxptr mode when transactions start */
  52 int ltdb_index_transaction_start(struct ldb_module *module)
  53 {
  54         struct ltdb_private *ltdb = talloc_get_type(ldb_module_get_private(module), struct ltdb_private);
  55         ltdb->idxptr = talloc_zero(ltdb, struct ltdb_idxptr);
  56         return LDB_SUCCESS;
  57 }
  58
  59 /* compare two DN entries in a dn_list. Take account of possible
  60  * differences in string termination */
  61 static int dn_list_cmp(const struct ldb_val *v1, const struct ldb_val *v2)
  62 {
  63         int ret = strncmp((char *)v1->data, (char *)v2->data, v1->length);
  64         if (ret != 0) return ret;
  65         if (v2->length > v1->length && v2->data[v1->length] != 0) {
  66                 return 1;
  67         }
  68         return 0;
  69 }
  70
  71
  72 /*
  73   find a entry in a dn_list, using a ldb_val. Uses a case sensitive
  74   comparison with the dn returns -1 if not found
  75  */
  76 static int ltdb_dn_list_find_val(const struct dn_list *list, const struct ldb_val *v)
  77 {
  78         int i;
  79         for (i=0; i<list->count; i++) {
  80                 if (dn_list_cmp(&list->dn[i], v) == 0) return i;
  81         }
  82         return -1;
  83 }
  84
  85 /*
  86   find a entry in a dn_list. Uses a case sensitive comparison with the dn
  87   returns -1 if not found
  88  */
  89 static int ltdb_dn_list_find_str(struct dn_list *list, const char *dn)
  90 {
  91         struct ldb_val v;
  92         v.data = discard_const_p(unsigned char, dn);
  93         v.length = strlen(dn);
  94         return ltdb_dn_list_find_val(list, &v);
  95 }
  96
  97 static struct dn_list *ltdb_index_idxptr(struct ldb_module *module, TDB_DATA rec, bool check_parent)
  98 {
  99         struct dn_list *list;
 100         if (rec.dsize != sizeof(void *)) {
 101                 ldb_asprintf_errstring(ldb_module_get_ctx(module),
 102                                        "Bad data size for idxptr %u", (unsigned)rec.dsize);
 103                 return NULL;
 104         }
 105
 106         list = talloc_get_type(*(struct dn_list **)rec.dptr, struct dn_list);
 107         if (list == NULL) {
 108                 ldb_asprintf_errstring(ldb_module_get_ctx(module),
 109                                        "Bad type '%s' for idxptr",
 110                                        talloc_get_name(*(struct dn_list **)rec.dptr));
 111                 return NULL;
 112         }
 113         if (check_parent && list->dn && talloc_parent(list->dn) != list) {
 114                 ldb_asprintf_errstring(ldb_module_get_ctx(module),
 115                                        "Bad parent '%s' for idxptr",
 116                                        talloc_get_name(talloc_parent(list->dn)));
 117                 return NULL;
 118         }
 119         return list;
 120 }
 121
 122 /*
 123   return the @IDX list in an index entry for a dn as a
 124   struct dn_list
 125  */
 126 static int ltdb_dn_list_load(struct ldb_module *module,
 127                              struct ldb_dn *dn, struct dn_list *list)
 128 {
 129         struct ldb_message *msg;
 130         int ret;
 131         struct ldb_message_element *el;
 132         struct ltdb_private *ltdb = talloc_get_type(ldb_module_get_private(module), struct ltdb_private);
 133         TDB_DATA rec;
 134         struct dn_list *list2;
 135         TDB_DATA key;
 136
 137         list->dn = NULL;
 138         list->count = 0;
 139
 140         /* see if we have any in-memory index entries */
 141         if (ltdb->idxptr == NULL ||
 142             ltdb->idxptr->itdb == NULL) {
 143                 goto normal_index;
 144         }
 145
 146         key.dptr = discard_const_p(unsigned char, ldb_dn_get_linearized(dn));
 147         key.dsize = strlen((char *)key.dptr);
 148
 149         rec = tdb_fetch(ltdb->idxptr->itdb, key);
 150         if (rec.dptr == NULL) {
 151                 goto normal_index;
 152         }
 153
 154         /* we've found an in-memory index entry */
 155         list2 = ltdb_index_idxptr(module, rec, true);
 156         if (list2 == NULL) {
 157                 free(rec.dptr);
 158                 return LDB_ERR_OPERATIONS_ERROR;
 159         }
 160         free(rec.dptr);
 161
 162         *list = *list2;
 163         return LDB_SUCCESS;
 164
 165 normal_index:
 166         msg = ldb_msg_new(list);
 167         if (msg == NULL) {
 168                 return LDB_ERR_OPERATIONS_ERROR;
 169         }
 170
 171         ret = ltdb_search_dn1(module, dn, msg);
 172         if (ret != LDB_SUCCESS) {
 173                 return ret;
 174         }
 175
 176         /* TODO: check indexing version number */
 177
 178         el = ldb_msg_find_element(msg, LTDB_IDX);
 179         if (!el) {
 180                 talloc_free(msg);
 181                 return LDB_SUCCESS;
 182         }
 183
 184         /* we avoid copying the strings by stealing the list */
 185         list->dn = talloc_steal(list, el->values);
 186         list->count = el->num_values;
 187
 188         return LDB_SUCCESS;
 189 }
 190
 191
 192 /*
 193   save a dn_list into a full @IDX style record
 194  */
 195 static int ltdb_dn_list_store_full(struct ldb_module *module, struct ldb_dn *dn,
 196                                    struct dn_list *list)
 197 {
 198         struct ldb_message *msg;
 199         int ret;
 200
 201         if (list->count == 0) {
 202                 ret = ltdb_delete_noindex(module, dn);
 203                 if (ret == LDB_ERR_NO_SUCH_OBJECT) {
 204                         return LDB_SUCCESS;
 205                 }
 206                 return ret;
 207         }
 208
 209         msg = ldb_msg_new(module);
 210         if (!msg) {
 211                 ldb_module_oom(module);
 212                 return LDB_ERR_OPERATIONS_ERROR;
 213         }
 214
 215         ret = ldb_msg_add_fmt(msg, LTDB_IDXVERSION, "%u", LTDB_INDEXING_VERSION);
 216         if (ret != LDB_SUCCESS) {
 217                 talloc_free(msg);
 218                 ldb_module_oom(module);
 219                 return LDB_ERR_OPERATIONS_ERROR;
 220         }
 221
 222         msg->dn = dn;
 223         if (list->count > 0) {
 224                 struct ldb_message_element *el;
 225
 226                 ret = ldb_msg_add_empty(msg, LTDB_IDX, LDB_FLAG_MOD_ADD, &el);
 227                 if (ret != LDB_SUCCESS) {
 228                         ldb_module_oom(module);
 229                         talloc_free(msg);
 230                         return LDB_ERR_OPERATIONS_ERROR;
 231                 }
 232                 el->values = list->dn;
 233                 el->num_values = list->count;
 234         }
 235
 236         ret = ltdb_store(module, msg, TDB_REPLACE);
 237         talloc_free(msg);
 238         return ret;
 239 }
 240
 241 /*
 242   save a dn_list into the database, in either @IDX or internal format
 243  */
 244 static int ltdb_dn_list_store(struct ldb_module *module, struct ldb_dn *dn,
 245                               struct dn_list *list)
 246 {
 247         struct ltdb_private *ltdb = talloc_get_type(ldb_module_get_private(module), struct ltdb_private);
 248         TDB_DATA rec, key;
 249         int ret;
 250         struct dn_list *list2;
 251
 252         if (ltdb->idxptr == NULL) {
 253                 return ltdb_dn_list_store_full(module, dn, list);
 254         }
 255
 256         if (ltdb->idxptr->itdb == NULL) {
 257                 ltdb->idxptr->itdb = tdb_open(NULL, 1000, TDB_INTERNAL, O_RDWR, 0);
 258                 if (ltdb->idxptr->itdb == NULL) {
 259                         return LDB_ERR_OPERATIONS_ERROR;
 260                 }
 261         }
 262
 263         key.dptr = discard_const_p(unsigned char, ldb_dn_get_linearized(dn));
 264         key.dsize = strlen((char *)key.dptr);
 265
 266         rec = tdb_fetch(ltdb->idxptr->itdb, key);
 267         if (rec.dptr != NULL) {
 268                 list2 = ltdb_index_idxptr(module, rec, false);
 269                 if (list2 == NULL) {
 270                         free(rec.dptr);
 271                         return LDB_ERR_OPERATIONS_ERROR;
 272                 }
 273                 free(rec.dptr);
 274                 list2->dn = talloc_steal(list2, list->dn);
 275                 list2->count = list->count;
 276                 return LDB_SUCCESS;
 277         }
 278
 279         list2 = talloc(ltdb->idxptr, struct dn_list);
 280         if (list2 == NULL) {
 281                 return LDB_ERR_OPERATIONS_ERROR;
 282         }
 283         list2->dn = talloc_steal(list2, list->dn);
 284         list2->count = list->count;
 285
 286         rec.dptr = (uint8_t *)&list2;
 287         rec.dsize = sizeof(void *);
 288
 289         ret = tdb_store(ltdb->idxptr->itdb, key, rec, TDB_INSERT);
 290         if (ret == -1) {
 291                 return ltdb_err_map(tdb_error(ltdb->idxptr->itdb));
 292         }
 293         return LDB_SUCCESS;
 294 }
 295
 296 /*
 297   traverse function for storing the in-memory index entries on disk
 298  */
 299 static int ltdb_index_traverse_store(struct tdb_context *tdb, TDB_DATA key, TDB_DATA data, void *state)
 300 {
 301         struct ldb_module *module = state;
 302         struct ltdb_private *ltdb = talloc_get_type(ldb_module_get_private(module), struct ltdb_private);
 303         struct ldb_dn *dn;
 304         struct ldb_context *ldb = ldb_module_get_ctx(module);
 305         struct ldb_val v;
 306         struct dn_list *list;
 307
 308         list = ltdb_index_idxptr(module, data, true);
 309         if (list == NULL) {
 310                 ltdb->idxptr->error = LDB_ERR_OPERATIONS_ERROR;
 311                 return -1;
 312         }
 313
 314         v.data = key.dptr;
 315         v.length = strnlen((char *)key.dptr, key.dsize);
 316
 317         dn = ldb_dn_from_ldb_val(module, ldb, &v);
 318         if (dn == NULL) {
 319                 ltdb->idxptr->error = LDB_ERR_OPERATIONS_ERROR;
 320                 return -1;
 321         }
 322
 323         ltdb->idxptr->error = ltdb_dn_list_store_full(module, dn, list);
 324         talloc_free(dn);
 325         if (ltdb->idxptr->error != 0) {
 326                 return -1;
 327         }
 328         return 0;
 329 }
 330
 331 /* cleanup the idxptr mode when transaction commits */
 332 int ltdb_index_transaction_commit(struct ldb_module *module)
 333 {
 334         struct ltdb_private *ltdb = talloc_get_type(ldb_module_get_private(module), struct ltdb_private);
 335         int ret;
 336
 337         if (ltdb->idxptr->itdb) {
 338                 tdb_traverse(ltdb->idxptr->itdb, ltdb_index_traverse_store, module);
 339                 tdb_close(ltdb->idxptr->itdb);
 340         }
 341
 342         ret = ltdb->idxptr->error;
 343
 344         if (ret != LDB_SUCCESS) {
 345                 struct ldb_context *ldb = ldb_module_get_ctx(module);
 346                 ldb_asprintf_errstring(ldb, "Failed to store index records in transaction commit");
 347         }
 348
 349         talloc_free(ltdb->idxptr);
 350         ltdb->idxptr = NULL;
 351         return ret;
 352 }
 353
 354 /* cleanup the idxptr mode when transaction cancels */
 355 int ltdb_index_transaction_cancel(struct ldb_module *module)
 356 {
 357         struct ltdb_private *ltdb = talloc_get_type(ldb_module_get_private(module), struct ltdb_private);
 358         if (ltdb->idxptr && ltdb->idxptr->itdb) {
 359                 tdb_close(ltdb->idxptr->itdb);
 360         }
 361         talloc_free(ltdb->idxptr);
 362         ltdb->idxptr = NULL;
 363         return LDB_SUCCESS;
 364 }
 365
 366
 367 /*
 368   return the dn key to be used for an index
 369   the caller is responsible for freeing
 370 */
 371 static struct ldb_dn *ltdb_index_key(struct ldb_context *ldb,
 372                                      const char *attr, const struct ldb_val *value,
 373                                      const struct ldb_schema_attribute **ap)
 374 {
 375         struct ldb_dn *ret;
 376         struct ldb_val v;
 377         const struct ldb_schema_attribute *a;
 378         char *attr_folded;
 379         int r;
 380
 381         attr_folded = ldb_attr_casefold(ldb, attr);
 382         if (!attr_folded) {
 383                 return NULL;
 384         }
 385
 386         a = ldb_schema_attribute_by_name(ldb, attr);
 387         if (ap) {
 388                 *ap = a;
 389         }
 390         r = a->syntax->canonicalise_fn(ldb, ldb, value, &v);
 391         if (r != LDB_SUCCESS) {
 392                 const char *errstr = ldb_errstring(ldb);
 393                 /* canonicalisation can be refused. For example,
 394                    a attribute that takes wildcards will refuse to canonicalise
 395                    if the value contains a wildcard */
 396                 ldb_asprintf_errstring(ldb, "Failed to create index key for attribute '%s':%s%s%s",
 397                                        attr, ldb_strerror(r), (errstr?":":""), (errstr?errstr:""));
 398                 talloc_free(attr_folded);
 399                 return NULL;
 400         }
 401         if (ldb_should_b64_encode(ldb, &v)) {
 402                 char *vstr = ldb_base64_encode(ldb, (char *)v.data, v.length);
 403                 if (!vstr) return NULL;
 404                 /* remove trailing '=' to make it a valid DN */
 405                 if (vstr[strlen(vstr)-1] == '=') {
 406                         vstr[strlen(vstr)-1] = 0;
 407                 }
 408                 ret = ldb_dn_new_fmt(ldb, ldb, "%s:%s::%s", LTDB_INDEX, attr_folded, vstr);
 409                 talloc_free(vstr);
 410         } else {
 411                 ret = ldb_dn_new_fmt(ldb, ldb, "%s:%s:%.*s", LTDB_INDEX, attr_folded, (int)v.length, (char *)v.data);
 412         }
 413
 414         if (v.data != value->data) {
 415                 talloc_free(v.data);
 416         }
 417         talloc_free(attr_folded);
 418
 419         return ret;
 420 }
 421
 422 /*
 423   see if a attribute value is in the list of indexed attributes
 424 */
 425 static bool ltdb_is_indexed(const struct ldb_message *index_list, const char *attr)
 426 {
 427         unsigned int i;
 428         struct ldb_message_element *el;
 429
 430         el = ldb_msg_find_element(index_list, LTDB_IDXATTR);
 431         if (el == NULL) {
 432                 return false;
 433         }
 434
 435         /* TODO: this is too expensive! At least use a binary search */
 436         for (i=0; i<el->num_values; i++) {
 437                 if (ldb_attr_cmp((char *)el->values[i].data, attr) == 0) {
 438                         return true;
 439                 }
 440         }
 441         return false;
 442 }
 443
 444 /*
 445   in the following logic functions, the return value is treated as
 446   follows:
 447
 448      LDB_SUCCESS: we found some matching index values
 449
 450      LDB_ERR_NO_SUCH_OBJECT: we know for sure that no object matches
 451
 452      LDB_ERR_OPERATIONS_ERROR: indexing could not answer the call,
 453                                we'll need a full search
 454  */
 455
 456 /*
 457   return a list of dn's that might match a simple indexed search (an
 458   equality search only)
 459  */
 460 static int ltdb_index_dn_simple(struct ldb_module *module,
 461                                 const struct ldb_parse_tree *tree,
 462                                 const struct ldb_message *index_list,
 463                                 struct dn_list *list)
 464 {
 465         struct ldb_context *ldb;
 466         struct ldb_dn *dn;
 467         int ret;
 468
 469         ldb = ldb_module_get_ctx(module);
 470
 471         list->count = 0;
 472         list->dn = NULL;
 473
 474         /* if the attribute isn't in the list of indexed attributes then
 475            this node needs a full search */
 476         if (!ltdb_is_indexed(index_list, tree->u.equality.attr)) {
 477                 return LDB_ERR_OPERATIONS_ERROR;
 478         }
 479
 480         /* the attribute is indexed. Pull the list of DNs that match the
 481            search criterion */
 482         dn = ltdb_index_key(ldb, tree->u.equality.attr, &tree->u.equality.value, NULL);
 483         if (!dn) return LDB_ERR_OPERATIONS_ERROR;
 484
 485         ret = ltdb_dn_list_load(module, dn, list);
 486         talloc_free(dn);
 487         return ret;
 488 }
 489
 490
 491 static bool list_union(struct ldb_context *, struct dn_list *, const struct dn_list *);
 492
 493 /*
 494   return a list of dn's that might match a leaf indexed search
 495  */
 496 static int ltdb_index_dn_leaf(struct ldb_module *module,
 497                               const struct ldb_parse_tree *tree,
 498                               const struct ldb_message *index_list,
 499                               struct dn_list *list)
 500 {
 501         if (ldb_attr_dn(tree->u.equality.attr) == 0) {
 502                 list->dn = talloc_array(list, struct ldb_val, 1);
 503                 if (list->dn == NULL) {
 504                         ldb_module_oom(module);
 505                         return LDB_ERR_OPERATIONS_ERROR;
 506                 }
 507                 list->dn[0] = tree->u.equality.value;
 508                 list->count = 1;
 509                 return LDB_SUCCESS;
 510         }
 511         return ltdb_index_dn_simple(module, tree, index_list, list);
 512 }
 513
 514
 515 /*
 516   list intersection
 517   list = list & list2
 518 */
 519 static bool list_intersect(struct ldb_context *ldb,
 520                            struct dn_list *list, const struct dn_list *list2)
 521 {
 522         struct dn_list *list3;
 523         unsigned int i;
 524
 525         if (list->count == 0) {
 526                 /* 0 & X == 0 */
 527                 return true;
 528         }
 529         if (list2->count == 0) {
 530                 /* X & 0 == 0 */
 531                 list->count = 0;
 532                 list->dn = NULL;
 533                 return true;
 534         }
 535
 536         /* the indexing code is allowed to return a longer list than
 537            what really matches, as all results are filtered by the
 538            full expression at the end - this shortcut avoids a lot of
 539            work in some cases */
 540         if (list->count < 2 && list2->count > 10) {
 541                 return true;
 542         }
 543         if (list2->count < 2 && list->count > 10) {
 544                 list->count = list2->count;
 545                 list->dn = list2->dn;
 546                 /* note that list2 may not be the parent of list2->dn,
 547                    as list2->dn may be owned by ltdb->idxptr. In that
 548                    case we expect this reparent call to fail, which is
 549                    OK */
 550                 talloc_reparent(list2, list, list2->dn);
 551                 return true;
 552         }
 553
 554         list3 = talloc_zero(list, struct dn_list);
 555         if (list3 == NULL) {
 556                 return false;
 557         }
 558
 559         list3->dn = talloc_array(list3, struct ldb_val, list->count);
 560         if (!list3->dn) {
 561                 talloc_free(list3);
 562                 return false;
 563         }
 564         list3->count = 0;
 565
 566         for (i=0;i<list->count;i++) {
 567                 if (ltdb_dn_list_find_val(list2, &list->dn[i]) != -1) {
 568                         list3->dn[list3->count] = list->dn[i];
 569                         list3->count++;
 570                 }
 571         }
 572
 573         list->dn = talloc_steal(list, list3->dn);
 574         list->count = list3->count;
 575         talloc_free(list3);
 576
 577         return true;
 578 }
 579
 580
 581 /*
 582   list union
 583   list = list | list2
 584 */
 585 static bool list_union(struct ldb_context *ldb,
 586                        struct dn_list *list, const struct dn_list *list2)
 587 {
 588         struct ldb_val *dn3;
 589
 590         if (list2->count == 0) {
 591                 /* X | 0 == X */
 592                 return true;
 593         }
 594
 595         if (list->count == 0) {
 596                 /* 0 | X == X */
 597                 list->count = list2->count;
 598                 list->dn = list2->dn;
 599                 /* note that list2 may not be the parent of list2->dn,
 600                    as list2->dn may be owned by ltdb->idxptr. In that
 601                    case we expect this reparent call to fail, which is
 602                    OK */
 603                 talloc_reparent(list2, list, list2->dn);
 604                 return true;
 605         }
 606
 607         dn3 = talloc_array(list, struct ldb_val, list->count + list2->count);
 608         if (!dn3) {
 609                 ldb_oom(ldb);
 610                 return false;
 611         }
 612
 613         /* we allow for duplicates here, and get rid of them later */
 614         memcpy(dn3, list->dn, sizeof(list->dn[0])*list->count);
 615         memcpy(dn3+list->count, list2->dn, sizeof(list2->dn[0])*list2->count);
 616
 617         list->dn = dn3;
 618         list->count += list2->count;
 619
 620         return true;
 621 }
 622
 623 static int ltdb_index_dn(struct ldb_module *module,
 624                          const struct ldb_parse_tree *tree,
 625                          const struct ldb_message *index_list,
 626                          struct dn_list *list);
 627
 628
 629 /*
 630   process an OR list (a union)
 631  */
 632 static int ltdb_index_dn_or(struct ldb_module *module,
 633                             const struct ldb_parse_tree *tree,
 634                             const struct ldb_message *index_list,
 635                             struct dn_list *list)
 636 {
 637         struct ldb_context *ldb;
 638         unsigned int i;
 639
 640         ldb = ldb_module_get_ctx(module);
 641
 642         list->dn = NULL;
 643         list->count = 0;
 644
 645         for (i=0; i<tree->u.list.num_elements; i++) {
 646                 struct dn_list *list2;
 647                 int ret;
 648
 649                 list2 = talloc_zero(list, struct dn_list);
 650                 if (list2 == NULL) {
 651                         return LDB_ERR_OPERATIONS_ERROR;
 652                 }
 653
 654                 ret = ltdb_index_dn(module, tree->u.list.elements[i], index_list, list2);
 655
 656                 if (ret == LDB_ERR_NO_SUCH_OBJECT) {
 657                         /* X || 0 == X */
 658                         talloc_free(list2);
 659                         continue;
 660                 }
 661
 662                 if (ret != LDB_SUCCESS) {
 663                         /* X || * == * */
 664                         talloc_free(list2);
 665                         return ret;
 666                 }
 667
 668                 if (!list_union(ldb, list, list2)) {
 669                         talloc_free(list2);
 670                         return LDB_ERR_OPERATIONS_ERROR;
 671                 }
 672         }
 673
 674         if (list->count == 0) {
 675                 return LDB_ERR_NO_SUCH_OBJECT;
 676         }
 677
 678         return LDB_SUCCESS;
 679 }
 680
 681
 682 /*
 683   NOT an index results
 684  */
 685 static int ltdb_index_dn_not(struct ldb_module *module,
 686                              const struct ldb_parse_tree *tree,
 687                              const struct ldb_message *index_list,
 688                              struct dn_list *list)
 689 {
 690         /* the only way to do an indexed not would be if we could
 691            negate the not via another not or if we knew the total
 692            number of database elements so we could know that the
 693            existing expression covered the whole database.
 694
 695            instead, we just give up, and rely on a full index scan
 696            (unless an outer & manages to reduce the list)
 697         */
 698         return LDB_ERR_OPERATIONS_ERROR;
 699 }
 700
 701
 702 static bool ltdb_index_unique(struct ldb_context *ldb,
 703                               const char *attr)
 704 {
 705         const struct ldb_schema_attribute *a;
 706         a = ldb_schema_attribute_by_name(ldb, attr);
 707         if (a->flags & LDB_ATTR_FLAG_UNIQUE_INDEX) {
 708                 return true;
 709         }
 710         return false;
 711 }
 712
 713 /*
 714   process an AND expression (intersection)
 715  */
 716 static int ltdb_index_dn_and(struct ldb_module *module,
 717                              const struct ldb_parse_tree *tree,
 718                              const struct ldb_message *index_list,
 719                              struct dn_list *list)
 720 {
 721         struct ldb_context *ldb;
 722         unsigned int i;
 723         bool found;
 724
 725         ldb = ldb_module_get_ctx(module);
 726
 727         list->dn = NULL;
 728         list->count = 0;
 729
 730         /* in the first pass we only look for unique simple
 731            equality tests, in the hope of avoiding having to look
 732            at any others */
 733         for (i=0; i<tree->u.list.num_elements; i++) {
 734                 const struct ldb_parse_tree *subtree = tree->u.list.elements[i];
 735                 int ret;
 736
 737                 if (subtree->operation != LDB_OP_EQUALITY ||
 738                     !ltdb_index_unique(ldb, subtree->u.equality.attr)) {
 739                         continue;
 740                 }
 741
 742                 ret = ltdb_index_dn(module, subtree, index_list, list);
 743                 if (ret == LDB_ERR_NO_SUCH_OBJECT) {
 744                         /* 0 && X == 0 */
 745                         return LDB_ERR_NO_SUCH_OBJECT;
 746                 }
 747                 if (ret == LDB_SUCCESS) {
 748                         /* a unique index match means we can
 749                          * stop. Note that we don't care if we return
 750                          * a few too many objects, due to later
 751                          * filtering */
 752                         return LDB_SUCCESS;
 753                 }
 754         }
 755
 756         /* now do a full intersection */
 757         found = false;
 758
 759         for (i=0; i<tree->u.list.num_elements; i++) {
 760                 const struct ldb_parse_tree *subtree = tree->u.list.elements[i];
 761                 struct dn_list *list2;
 762                 int ret;
 763
 764                 list2 = talloc_zero(list, struct dn_list);
 765                 if (list2 == NULL) {
 766                         ldb_module_oom(module);
 767                         return LDB_ERR_OPERATIONS_ERROR;
 768                 }
 769
 770                 ret = ltdb_index_dn(module, subtree, index_list, list2);
 771
 772                 if (ret == LDB_ERR_NO_SUCH_OBJECT) {
 773                         /* X && 0 == 0 */
 774                         list->dn = NULL;
 775                         list->count = 0;
 776                         talloc_free(list2);
 777                         return LDB_ERR_NO_SUCH_OBJECT;
 778                 }
 779
 780                 if (ret != LDB_SUCCESS) {
 781                         /* this didn't adding anything */
 782                         talloc_free(list2);
 783                         continue;
 784                 }
 785
 786                 if (!found) {
 787                         talloc_reparent(list2, list, list->dn);
 788                         list->dn = list2->dn;
 789                         list->count = list2->count;
 790                         found = true;
 791                 } else if (!list_intersect(ldb, list, list2)) {
 792                         talloc_free(list2);
 793                         return LDB_ERR_OPERATIONS_ERROR;
 794                 }
 795
 796                 if (list->count == 0) {
 797                         list->dn = NULL;
 798                         return LDB_ERR_NO_SUCH_OBJECT;
 799                 }
 800
 801                 if (list->count < 2) {
 802                         /* it isn't worth loading the next part of the tree */
 803                         return LDB_SUCCESS;
 804                 }
 805         }
 806
 807         if (!found) {
 808                 /* none of the attributes were indexed */
 809                 return LDB_ERR_OPERATIONS_ERROR;
 810         }
 811
 812         return LDB_SUCCESS;
 813 }
 814
 815 /*
 816   return a list of matching objects using a one-level index
 817  */
 818 static int ltdb_index_dn_one(struct ldb_module *module,
 819                              struct ldb_dn *parent_dn,
 820                              struct dn_list *list)
 821 {
 822         struct ldb_context *ldb;
 823         struct ldb_dn *key;
 824         struct ldb_val val;
 825         int ret;
 826
 827         ldb = ldb_module_get_ctx(module);
 828
 829         /* work out the index key from the parent DN */
 830         val.data = (uint8_t *)((uintptr_t)ldb_dn_get_casefold(parent_dn));
 831         val.length = strlen((char *)val.data);
 832         key = ltdb_index_key(ldb, LTDB_IDXONE, &val, NULL);
 833         if (!key) {
 834                 ldb_oom(ldb);
 835                 return LDB_ERR_OPERATIONS_ERROR;
 836         }
 837
 838         ret = ltdb_dn_list_load(module, key, list);
 839         talloc_free(key);
 840         if (ret != LDB_SUCCESS) {
 841                 return ret;
 842         }
 843
 844         if (list->count == 0) {
 845                 return LDB_ERR_NO_SUCH_OBJECT;
 846         }
 847
 848         return LDB_SUCCESS;
 849 }
 850
 851 /*
 852   return a list of dn's that might match a indexed search or
 853   an error. return LDB_ERR_NO_SUCH_OBJECT for no matches, or LDB_SUCCESS for matches
 854  */
 855 static int ltdb_index_dn(struct ldb_module *module,
 856                          const struct ldb_parse_tree *tree,
 857                          const struct ldb_message *index_list,
 858                          struct dn_list *list)
 859 {
 860         int ret = LDB_ERR_OPERATIONS_ERROR;
 861
 862         switch (tree->operation) {
 863         case LDB_OP_AND:
 864                 ret = ltdb_index_dn_and(module, tree, index_list, list);
 865                 break;
 866
 867         case LDB_OP_OR:
 868                 ret = ltdb_index_dn_or(module, tree, index_list, list);
 869                 break;
 870
 871         case LDB_OP_NOT:
 872                 ret = ltdb_index_dn_not(module, tree, index_list, list);
 873                 break;
 874
 875         case LDB_OP_EQUALITY:
 876                 ret = ltdb_index_dn_leaf(module, tree, index_list, list);
 877                 break;
 878
 879         case LDB_OP_SUBSTRING:
 880         case LDB_OP_GREATER:
 881         case LDB_OP_LESS:
 882         case LDB_OP_PRESENT:
 883         case LDB_OP_APPROX:
 884         case LDB_OP_EXTENDED:
 885                 /* we can't index with fancy bitops yet */
 886                 ret = LDB_ERR_OPERATIONS_ERROR;
 887                 break;
 888         }
 889
 890         return ret;
 891 }
 892
 893 /*
 894   filter a candidate dn_list from an indexed search into a set of results
 895   extracting just the given attributes
 896 */
 897 static int ltdb_index_filter(const struct dn_list *dn_list,
 898                              struct ltdb_context *ac,
 899                              uint32_t *match_count)
 900 {
 901         struct ldb_context *ldb;
 902         struct ldb_message *msg;
 903         unsigned int i;
 904
 905         ldb = ldb_module_get_ctx(ac->module);
 906
 907         for (i = 0; i < dn_list->count; i++) {
 908                 struct ldb_dn *dn;
 909                 int ret;
 910
 911                 msg = ldb_msg_new(ac);
 912                 if (!msg) {
 913                         return LDB_ERR_OPERATIONS_ERROR;
 914                 }
 915
 916                 dn = ldb_dn_from_ldb_val(msg, ldb, &dn_list->dn[i]);
 917                 if (dn == NULL) {
 918                         talloc_free(msg);
 919                         return LDB_ERR_OPERATIONS_ERROR;
 920                 }
 921
 922                 ret = ltdb_search_dn1(ac->module, dn, msg);
 923                 talloc_free(dn);
 924                 if (ret == LDB_ERR_NO_SUCH_OBJECT) {
 925                         /* the record has disappeared? yes, this can happen */
 926                         talloc_free(msg);
 927                         continue;
 928                 }
 929
 930                 if (ret != LDB_SUCCESS && ret != LDB_ERR_NO_SUCH_OBJECT) {
 931                         /* an internal error */
 932                         talloc_free(msg);
 933                         return LDB_ERR_OPERATIONS_ERROR;
 934                 }
 935
 936                 if (!ldb_match_msg(ldb, msg,
 937                                    ac->tree, ac->base, ac->scope)) {
 938                         talloc_free(msg);
 939                         continue;
 940                 }
 941
 942                 /* filter the attributes that the user wants */
 943                 ret = ltdb_filter_attrs(msg, ac->attrs);
 944
 945                 if (ret == -1) {
 946                         talloc_free(msg);
 947                         return LDB_ERR_OPERATIONS_ERROR;
 948                 }
 949
 950                 ret = ldb_module_send_entry(ac->req, msg, NULL);
 951                 if (ret != LDB_SUCCESS) {
 952                         ac->request_terminated = true;
 953                         return ret;
 954                 }
 955
 956                 (*match_count)++;
 957         }
 958
 959         return LDB_SUCCESS;
 960 }
 961
 962 /*
 963   remove any duplicated entries in a indexed result
 964  */
 965 static void ltdb_dn_list_remove_duplicates(struct dn_list *list)
 966 {
 967         int i, new_count;
 968
 969         if (list->count < 2) {
 970                 return;
 971         }
 972
 973         qsort(list->dn, list->count, sizeof(struct ldb_val), (comparison_fn_t) dn_list_cmp);
 974
 975         new_count = 1;
 976         for (i=1; i<list->count; i++) {
 977                 if (dn_list_cmp(&list->dn[i], &list->dn[new_count-1]) != 0) {
 978                         if (new_count != i) {
 979                                 list->dn[new_count] = list->dn[i];
 980                         }
 981                         new_count++;
 982                 }
 983         }
 984
 985         list->count = new_count;
 986 }
 987
 988 /*
 989   search the database with a LDAP-like expression using indexes
 990   returns -1 if an indexed search is not possible, in which
 991   case the caller should call ltdb_search_full()
 992 */
 993 int ltdb_search_indexed(struct ltdb_context *ac, uint32_t *match_count)
 994 {
 995         struct ltdb_private *ltdb = talloc_get_type(ldb_module_get_private(ac->module), struct ltdb_private);
 996         struct dn_list *dn_list;
 997         int ret;
 998
 999         /* see if indexing is enabled */
1000         if (!ltdb->cache->attribute_indexes &&
1001             !ltdb->cache->one_level_indexes &&
1002             ac->scope != LDB_SCOPE_BASE) {
1003                 /* fallback to a full search */
1004                 return LDB_ERR_OPERATIONS_ERROR;
1005         }
1006
1007         dn_list = talloc_zero(ac, struct dn_list);
1008         if (dn_list == NULL) {
1009                 ldb_module_oom(ac->module);
1010                 return LDB_ERR_OPERATIONS_ERROR;
1011         }
1012
1013         switch (ac->scope) {
1014         case LDB_SCOPE_BASE:
1015                 dn_list->dn = talloc_array(dn_list, struct ldb_val, 1);
1016                 if (dn_list->dn == NULL) {
1017                         ldb_module_oom(ac->module);
1018                         talloc_free(dn_list);
1019                         return LDB_ERR_OPERATIONS_ERROR;
1020                 }
1021                 dn_list->dn[0].data = discard_const_p(unsigned char, ldb_dn_get_linearized(ac->base));
1022                 if (dn_list->dn[0].data == NULL) {
1023                         ldb_module_oom(ac->module);
1024                         talloc_free(dn_list);
1025                         return LDB_ERR_OPERATIONS_ERROR;
1026                 }
1027                 dn_list->dn[0].length = strlen((char *)dn_list->dn[0].data);
1028                 dn_list->count = 1;
1029                 break;
1030
1031         case LDB_SCOPE_ONELEVEL:
1032                 if (!ltdb->cache->one_level_indexes) {
1033                         talloc_free(dn_list);
1034                         return LDB_ERR_OPERATIONS_ERROR;
1035                 }
1036                 ret = ltdb_index_dn_one(ac->module, ac->base, dn_list);
1037                 if (ret != LDB_SUCCESS) {
1038                         talloc_free(dn_list);
1039                         return ret;
1040                 }
1041                 break;
1042
1043         case LDB_SCOPE_SUBTREE:
1044         case LDB_SCOPE_DEFAULT:
1045                 if (!ltdb->cache->attribute_indexes) {
1046                         talloc_free(dn_list);
1047                         return LDB_ERR_OPERATIONS_ERROR;
1048                 }
1049                 ret = ltdb_index_dn(ac->module, ac->tree, ltdb->cache->indexlist, dn_list);
1050                 if (ret != LDB_SUCCESS) {
1051                         talloc_free(dn_list);
1052                         return ret;
1053                 }
1054                 ltdb_dn_list_remove_duplicates(dn_list);
1055                 break;
1056         }
1057
1058         ret = ltdb_index_filter(dn_list, ac, match_count);
1059         talloc_free(dn_list);
1060         return ret;
1061 }
1062
1063 /*
1064   add an index entry for one message element
1065 */
1066 static int ltdb_index_add1(struct ldb_module *module, const char *dn,
1067                            struct ldb_message_element *el, int v_idx)
1068 {
1069         struct ldb_context *ldb;
1070         struct ldb_dn *dn_key;
1071         int ret;
1072         const struct ldb_schema_attribute *a;
1073         struct dn_list *list;
1074         unsigned alloc_len;
1075
1076         ldb = ldb_module_get_ctx(module);
1077
1078         list = talloc_zero(module, struct dn_list);
1079         if (list == NULL) {
1080                 return LDB_ERR_OPERATIONS_ERROR;
1081         }
1082
1083         dn_key = ltdb_index_key(ldb, el->name, &el->values[v_idx], &a);
1084         if (!dn_key) {
1085                 talloc_free(list);
1086                 return LDB_ERR_OPERATIONS_ERROR;
1087         }
1088         talloc_steal(list, dn_key);
1089
1090         ret = ltdb_dn_list_load(module, dn_key, list);
1091         if (ret != LDB_SUCCESS && ret != LDB_ERR_NO_SUCH_OBJECT) {
1092                 talloc_free(list);
1093                 return ret;
1094         }
1095
1096         if (ltdb_dn_list_find_str(list, dn) != -1) {
1097                 talloc_free(list);
1098                 return LDB_SUCCESS;
1099         }
1100
1101         if (list->count > 0 &&
1102             a->flags & LDB_ATTR_FLAG_UNIQUE_INDEX) {
1103                 talloc_free(list);
1104                 return LDB_ERR_ENTRY_ALREADY_EXISTS;
1105         }
1106
1107         /* overallocate the list a bit, to reduce the number of
1108          * realloc trigered copies */
1109         alloc_len = ((list->count+1)+7) & ~7;
1110         list->dn = talloc_realloc(list, list->dn, struct ldb_val, alloc_len);
1111         if (list->dn == NULL) {
1112                 talloc_free(list);
1113                 return LDB_ERR_OPERATIONS_ERROR;
1114         }
1115         list->dn[list->count].data = (uint8_t *)talloc_strdup(list->dn, dn);
1116         list->dn[list->count].length = strlen(dn);
1117         list->count++;
1118
1119         ret = ltdb_dn_list_store(module, dn_key, list);
1120
1121         talloc_free(list);
1122
1123         return ret;
1124 }
1125
1126 /*
1127   add index entries for one elements in a message
1128  */
1129 static int ltdb_index_add_el(struct ldb_module *module, const char *dn,
1130                              struct ldb_message_element *el)
1131 {
1132         unsigned int i;
1133         for (i = 0; i < el->num_values; i++) {
1134                 int ret = ltdb_index_add1(module, dn, el, i);
1135                 if (ret != LDB_SUCCESS) {
1136                         return ret;
1137                 }
1138         }
1139
1140         return LDB_SUCCESS;
1141 }
1142
1143 /*
1144   add index entries for all elements in a message
1145  */
1146 static int ltdb_index_add_all(struct ldb_module *module, const char *dn,
1147                               struct ldb_message_element *elements, int num_el)
1148 {
1149         struct ltdb_private *ltdb = talloc_get_type(ldb_module_get_private(module), struct ltdb_private);
1150         unsigned int i;
1151
1152         if (dn[0] == '@') {
1153                 return LDB_SUCCESS;
1154         }
1155
1156         if (ltdb->cache->indexlist->num_elements == 0) {
1157                 /* no indexed fields */
1158                 return LDB_SUCCESS;
1159         }
1160
1161         for (i = 0; i < num_el; i++) {
1162                 int ret;
1163                 if (!ltdb_is_indexed(ltdb->cache->indexlist, elements[i].name)) {
1164                         continue;
1165                 }
1166                 ret = ltdb_index_add_el(module, dn, &elements[i]);
1167                 if (ret != LDB_SUCCESS) {
1168                         return ret;
1169                 }
1170         }
1171
1172         return LDB_SUCCESS;
1173 }
1174
1175
1176 /*
1177   insert a one level index for a message
1178 */
1179 static int ltdb_index_onelevel(struct ldb_module *module, const struct ldb_message *msg, int add)
1180 {
1181         struct ltdb_private *ltdb = talloc_get_type(ldb_module_get_private(module), struct ltdb_private);
1182         struct ldb_message_element el;
1183         struct ldb_val val;
1184         struct ldb_dn *pdn;
1185         const char *dn;
1186         int ret;
1187
1188         /* We index for ONE Level only if requested */
1189         if (!ltdb->cache->one_level_indexes) {
1190                 return LDB_SUCCESS;
1191         }
1192
1193         pdn = ldb_dn_get_parent(module, msg->dn);
1194         if (pdn == NULL) {
1195                 return LDB_ERR_OPERATIONS_ERROR;
1196         }
1197
1198         dn = ldb_dn_get_linearized(msg->dn);
1199         if (dn == NULL) {
1200                 talloc_free(pdn);
1201                 return LDB_ERR_OPERATIONS_ERROR;
1202         }
1203
1204         val.data = (uint8_t *)((uintptr_t)ldb_dn_get_casefold(pdn));
1205         if (val.data == NULL) {
1206                 talloc_free(pdn);
1207                 return LDB_ERR_OPERATIONS_ERROR;
1208         }
1209
1210         val.length = strlen((char *)val.data);
1211         el.name = LTDB_IDXONE;
1212         el.values = &val;
1213         el.num_values = 1;
1214
1215         if (add) {
1216                 ret = ltdb_index_add1(module, dn, &el, 0);
1217         } else { /* delete */
1218                 ret = ltdb_index_del_value(module, dn, &el, 0);
1219         }
1220
1221         talloc_free(pdn);
1222
1223         return ret;
1224 }
1225
1226 /*
1227   add the index entries for a new element in a record
1228   The caller guarantees that these element values are not yet indexed
1229 */
1230 int ltdb_index_add_element(struct ldb_module *module, struct ldb_dn *dn,
1231                            struct ldb_message_element *el)
1232 {
1233         struct ltdb_private *ltdb = talloc_get_type(ldb_module_get_private(module), struct ltdb_private);
1234         if (ldb_dn_is_special(dn)) {
1235                 return LDB_SUCCESS;
1236         }
1237         if (!ltdb_is_indexed(ltdb->cache->indexlist, el->name)) {
1238                 return LDB_SUCCESS;
1239         }
1240         return ltdb_index_add_el(module, ldb_dn_get_linearized(dn), el);
1241 }
1242
1243 /*
1244   add the index entries for a new record
1245 */
1246 int ltdb_index_add_new(struct ldb_module *module, const struct ldb_message *msg)
1247 {
1248         const char *dn;
1249         int ret;
1250
1251         if (ldb_dn_is_special(msg->dn)) {
1252                 return LDB_SUCCESS;
1253         }
1254
1255         dn = ldb_dn_get_linearized(msg->dn);
1256         if (dn == NULL) {
1257                 return LDB_ERR_OPERATIONS_ERROR;
1258         }
1259
1260         ret = ltdb_index_add_all(module, dn, msg->elements, msg->num_elements);
1261         if (ret != LDB_SUCCESS) {
1262                 return ret;
1263         }
1264
1265         return ltdb_index_onelevel(module, msg, 1);
1266 }
1267
1268
1269 /*
1270   delete an index entry for one message element
1271 */
1272 int ltdb_index_del_value(struct ldb_module *module, const char *dn,
1273                          struct ldb_message_element *el, int v_idx)
1274 {
1275         struct ldb_context *ldb;
1276         struct ldb_dn *dn_key;
1277         int ret, i;
1278         struct dn_list *list;
1279
1280         ldb = ldb_module_get_ctx(module);
1281
1282         if (dn[0] == '@') {
1283                 return LDB_SUCCESS;
1284         }
1285
1286         dn_key = ltdb_index_key(ldb, el->name, &el->values[v_idx], NULL);
1287         if (!dn_key) {
1288                 return LDB_ERR_OPERATIONS_ERROR;
1289         }
1290
1291         list = talloc_zero(dn_key, struct dn_list);
1292         if (list == NULL) {
1293                 talloc_free(dn_key);
1294                 return LDB_ERR_OPERATIONS_ERROR;
1295         }
1296
1297         ret = ltdb_dn_list_load(module, dn_key, list);
1298         if (ret == LDB_ERR_NO_SUCH_OBJECT) {
1299                 /* it wasn't indexed. Did we have an earlier error? If we did then
1300                    its gone now */
1301                 talloc_free(dn_key);
1302                 return LDB_SUCCESS;
1303         }
1304
1305         if (ret != LDB_SUCCESS) {
1306                 talloc_free(dn_key);
1307                 return ret;
1308         }
1309
1310         i = ltdb_dn_list_find_str(list, dn);
1311         if (i == -1) {
1312                 /* nothing to delete */
1313                 talloc_free(dn_key);
1314                 return LDB_SUCCESS;
1315         }
1316
1317         if (i != list->count-1) {
1318                 memmove(&list->dn[i], &list->dn[i+1], sizeof(list->dn[0])*(list->count - (i+1)));
1319         }
1320         list->count--;
1321         list->dn = talloc_realloc(list, list->dn, struct ldb_val, list->count);
1322
1323         ret = ltdb_dn_list_store(module, dn_key, list);
1324
1325         talloc_free(dn_key);
1326
1327         return ret;
1328 }
1329
1330 /*
1331   delete the index entries for a element
1332   return -1 on failure
1333 */
1334 int ltdb_index_del_element(struct ldb_module *module, const char *dn, struct ldb_message_element *el)
1335 {
1336         struct ltdb_private *ltdb = talloc_get_type(ldb_module_get_private(module), struct ltdb_private);
1337         int ret;
1338         unsigned int i;
1339
1340         if (!ltdb->cache->attribute_indexes) {
1341                 /* no indexed fields */
1342                 return LDB_SUCCESS;
1343         }
1344
1345         if (dn[0] == '@') {
1346                 return LDB_SUCCESS;
1347         }
1348
1349         if (!ltdb_is_indexed(ltdb->cache->indexlist, el->name)) {
1350                 return LDB_SUCCESS;
1351         }
1352         for (i = 0; i < el->num_values; i++) {
1353                 ret = ltdb_index_del_value(module, dn, el, i);
1354                 if (ret != LDB_SUCCESS) {
1355                         return ret;
1356                 }
1357         }
1358
1359         return LDB_SUCCESS;
1360 }
1361
1362 /*
1363   delete the index entries for a record
1364   return -1 on failure
1365 */
1366 int ltdb_index_delete(struct ldb_module *module, const struct ldb_message *msg)
1367 {
1368         struct ltdb_private *ltdb = talloc_get_type(ldb_module_get_private(module), struct ltdb_private);
1369         int ret;
1370         const char *dn;
1371         unsigned int i;
1372
1373         if (ldb_dn_is_special(msg->dn)) {
1374                 return LDB_SUCCESS;
1375         }
1376
1377         ret = ltdb_index_onelevel(module, msg, 0);
1378         if (ret != LDB_SUCCESS) {
1379                 return ret;
1380         }
1381
1382         if (!ltdb->cache->attribute_indexes) {
1383                 /* no indexed fields */
1384                 return LDB_SUCCESS;
1385         }
1386
1387         dn = ldb_dn_get_linearized(msg->dn);
1388         if (dn == NULL) {
1389                 return LDB_ERR_OPERATIONS_ERROR;
1390         }
1391
1392         for (i = 0; i < msg->num_elements; i++) {
1393                 ret = ltdb_index_del_element(module, dn, &msg->elements[i]);
1394                 if (ret != LDB_SUCCESS) {
1395                         return ret;
1396                 }
1397         }
1398
1399         return LDB_SUCCESS;
1400 }
1401
1402
1403 /*
1404   traversal function that deletes all @INDEX records
1405 */
1406 static int delete_index(struct tdb_context *tdb, TDB_DATA key, TDB_DATA data, void *state)
1407 {
1408         struct ldb_module *module = state;
1409         struct ltdb_private *ltdb = talloc_get_type(ldb_module_get_private(module), struct ltdb_private);
1410         const char *dnstr = "DN=" LTDB_INDEX ":";
1411         struct dn_list list;
1412         struct ldb_dn *dn;
1413         struct ldb_val v;
1414         int ret;
1415
1416         if (strncmp((char *)key.dptr, dnstr, strlen(dnstr)) != 0) {
1417                 return 0;
1418         }
1419         /* we need to put a empty list in the internal tdb for this
1420          * index entry */
1421         list.dn = NULL;
1422         list.count = 0;
1423         v.data = key.dptr;
1424         v.length = strnlen((char *)key.dptr, key.dsize);
1425
1426         dn = ldb_dn_from_ldb_val(ltdb, ldb_module_get_ctx(module), &v);
1427         ret = ltdb_dn_list_store(module, dn, &list);
1428         if (ret != LDB_SUCCESS) {
1429                 ldb_asprintf_errstring(ldb_module_get_ctx(module),
1430                                        "Unable to store null index for %s\n",
1431                                        ldb_dn_get_linearized(dn));
1432                 talloc_free(dn);
1433                 return -1;
1434         }
1435         talloc_free(dn);
1436         return 0;
1437 }
1438
1439 /*
1440   traversal function that adds @INDEX records during a re index
1441 */
1442 static int re_index(struct tdb_context *tdb, TDB_DATA key, TDB_DATA data, void *state)
1443 {
1444         struct ldb_context *ldb;
1445         struct ldb_module *module = (struct ldb_module *)state;
1446         struct ldb_message *msg;
1447         const char *dn = NULL;
1448         int ret;
1449         TDB_DATA key2;
1450
1451         ldb = ldb_module_get_ctx(module);
1452
1453         if (strncmp((char *)key.dptr, "DN=@", 4) == 0 ||
1454             strncmp((char *)key.dptr, "DN=", 3) != 0) {
1455                 return 0;
1456         }
1457
1458         msg = ldb_msg_new(module);
1459         if (msg == NULL) {
1460                 return -1;
1461         }
1462
1463         ret = ltdb_unpack_data(module, &data, msg);
1464         if (ret != 0) {
1465                 ldb_debug(ldb, LDB_DEBUG_ERROR, "Invalid data for index %s\n",
1466                           ldb_dn_get_linearized(msg->dn));
1467                 talloc_free(msg);
1468                 return -1;
1469         }
1470
1471         /* check if the DN key has changed, perhaps due to the
1472            case insensitivity of an element changing */
1473         key2 = ltdb_key(module, msg->dn);
1474         if (key2.dptr == NULL) {
1475                 /* probably a corrupt record ... darn */
1476                 ldb_debug(ldb, LDB_DEBUG_ERROR, "Invalid DN in re_index: %s",
1477                                                         ldb_dn_get_linearized(msg->dn));
1478                 talloc_free(msg);
1479                 return 0;
1480         }
1481         if (strcmp((char *)key2.dptr, (char *)key.dptr) != 0) {
1482                 tdb_delete(tdb, key);
1483                 tdb_store(tdb, key2, data, 0);
1484         }
1485         talloc_free(key2.dptr);
1486
1487         if (msg->dn == NULL) {
1488                 dn = (char *)key.dptr + 3;
1489         } else {
1490                 dn = ldb_dn_get_linearized(msg->dn);
1491         }
1492
1493         ret = ltdb_index_onelevel(module, msg, 1);
1494         if (ret != LDB_SUCCESS) {
1495                 ldb_debug(ldb, LDB_DEBUG_ERROR,
1496                         "Adding special ONE LEVEL index failed (%s)!",
1497                         ldb_dn_get_linearized(msg->dn));
1498                 talloc_free(msg);
1499                 return -1;
1500         }
1501
1502         ret = ltdb_index_add_all(module, dn, msg->elements, msg->num_elements);
1503
1504         talloc_free(msg);
1505
1506         if (ret != LDB_SUCCESS) return -1;
1507
1508         return 0;
1509 }
1510
1511 /*
1512   force a complete reindex of the database
1513 */
1514 int ltdb_reindex(struct ldb_module *module)
1515 {
1516         struct ltdb_private *ltdb = talloc_get_type(ldb_module_get_private(module), struct ltdb_private);
1517         int ret;
1518
1519         if (ltdb_cache_reload(module) != 0) {
1520                 return LDB_ERR_OPERATIONS_ERROR;
1521         }
1522
1523         /* first traverse the database deleting any @INDEX records by
1524          * putting NULL entries in the in-memory tdb
1525          */
1526         ret = tdb_traverse(ltdb->tdb, delete_index, module);
1527         if (ret == -1) {
1528                 return LDB_ERR_OPERATIONS_ERROR;
1529         }
1530
1531         /* if we don't have indexes we have nothing todo */
1532         if (ltdb->cache->indexlist->num_elements == 0) {
1533                 return LDB_SUCCESS;
1534         }
1535
1536         /* now traverse adding any indexes for normal LDB records */
1537         ret = tdb_traverse(ltdb->tdb, re_index, module);
1538         if (ret == -1) {
1539                 return LDB_ERR_OPERATIONS_ERROR;
1540         }
1541
1542         return LDB_SUCCESS;
1543 }