source/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/includes.h"
  37
  38 #include "ldb/ldb_tdb/ldb_tdb.h"
  39
  40 /*
  41   find an element in a list, using the given comparison function and
  42   assuming that the list is already sorted using comp_fn
  43
  44   return -1 if not found, or the index of the first occurance of needle if found
  45 */
  46 static int ldb_list_find(const void *needle,
  47                          const void *base, size_t nmemb, size_t size,
  48                          comparison_fn_t comp_fn)
  49 {
  50         const char *base_p = base;
  51         size_t min_i, max_i, test_i;
  52
  53         if (nmemb == 0) {
  54                 return -1;
  55         }
  56
  57         min_i = 0;
  58         max_i = nmemb-1;
  59
  60         while (min_i < max_i) {
  61                 int r;
  62
  63                 test_i = (min_i + max_i) / 2;
  64                 /* the following cast looks strange, but is
  65                  correct. The key to understanding it is that base_p
  66                  is a pointer to an array of pointers, so we have to
  67                  dereference it after casting to void **. The strange
  68                  const in the middle gives us the right type of pointer
  69                  after the dereference  (tridge) */
  70                 r = comp_fn(needle, *(void * const *)(base_p + (size * test_i)));
  71                 if (r == 0) {
  72                         /* scan back for first element */
  73                         while (test_i > 0 &&
  74                                comp_fn(needle, *(void * const *)(base_p + (size * (test_i-1)))) == 0) {
  75                                 test_i--;
  76                         }
  77                         return test_i;
  78                 }
  79                 if (r < 0) {
  80                         if (test_i == 0) {
  81                                 return -1;
  82                         }
  83                         max_i = test_i - 1;
  84                 }
  85                 if (r > 0) {
  86                         min_i = test_i + 1;
  87                 }
  88         }
  89
  90         if (comp_fn(needle, *(void * const *)(base_p + (size * min_i))) == 0) {
  91                 return min_i;
  92         }
  93
  94         return -1;
  95 }
  96
  97 struct dn_list {
  98         unsigned int count;
  99         char **dn;
 100 };
 101
 102 /*
 103   return the dn key to be used for an index
 104   caller frees
 105 */
 106 static struct ldb_dn *ldb_dn_key(struct ldb_context *ldb,
 107                         const char *attr, const struct ldb_val *value)
 108 {
 109         struct ldb_dn *ret;
 110         char *dn;
 111         struct ldb_val v;
 112         const struct ldb_attrib_handler *h;
 113         char *attr_folded;
 114
 115         attr_folded = ldb_attr_casefold(ldb, attr);
 116         if (!attr_folded) {
 117                 return NULL;
 118         }
 119
 120         h = ldb_attrib_handler(ldb, attr);
 121         if (h->canonicalise_fn(ldb, ldb, value, &v) != 0) {
 122                 /* canonicalisation can be refused. For example,
 123                    a attribute that takes wildcards will refuse to canonicalise
 124                    if the value contains a wildcard */
 125                 talloc_free(attr_folded);
 126                 return NULL;
 127         }
 128         if (ldb_should_b64_encode(&v)) {
 129                 char *vstr = ldb_base64_encode(ldb, (char *)v.data, v.length);
 130                 if (!vstr) return NULL;
 131                 dn = talloc_asprintf(ldb, "%s:%s::%s", LTDB_INDEX, attr_folded, vstr);
 132                 talloc_free(vstr);
 133                 if (v.data != value->data) {
 134                         talloc_free(v.data);
 135                 }
 136                 talloc_free(attr_folded);
 137                 if (dn == NULL) return NULL;
 138                 goto done;
 139         }
 140
 141         dn = talloc_asprintf(ldb, "%s:%s:%.*s",
 142                               LTDB_INDEX, attr_folded, (int)v.length, (char *)v.data);
 143
 144         if (v.data != value->data) {
 145                 talloc_free(v.data);
 146         }
 147         talloc_free(attr_folded);
 148
 149 done:
 150         ret = ldb_dn_explode(ldb, dn);
 151         talloc_free(dn);
 152         return ret;
 153 }
 154
 155 /*
 156   see if a attribute value is in the list of indexed attributes
 157 */
 158 static int ldb_msg_find_idx(const struct ldb_message *msg, const char *attr,
 159                             unsigned int *v_idx, const char *key)
 160 {
 161         unsigned int i, j;
 162         for (i=0;i<msg->num_elements;i++) {
 163                 if (ldb_attr_cmp(msg->elements[i].name, key) == 0) {
 164                         const struct ldb_message_element *el =
 165                                 &msg->elements[i];
 166                         for (j=0;j<el->num_values;j++) {
 167                                 if (ldb_attr_cmp((char *)el->values[j].data, attr) == 0) {
 168                                         if (v_idx) {
 169                                                 *v_idx = j;
 170                                         }
 171                                         return i;
 172                                 }
 173                         }
 174                 }
 175         }
 176         return -1;
 177 }
 178
 179 /* used in sorting dn lists */
 180 static int list_cmp(const char **s1, const char **s2)
 181 {
 182         return strcmp(*s1, *s2);
 183 }
 184
 185 /*
 186   return a list of dn's that might match a simple indexed search or
 187  */
 188 static int ltdb_index_dn_simple(struct ldb_module *module,
 189                                 const struct ldb_parse_tree *tree,
 190                                 const struct ldb_message *index_list,
 191                                 struct dn_list *list)
 192 {
 193         struct ldb_context *ldb = module->ldb;
 194         struct ldb_dn *dn;
 195         int ret;
 196         unsigned int i, j;
 197         struct ldb_message *msg;
 198
 199         list->count = 0;
 200         list->dn = NULL;
 201
 202         /* if the attribute isn't in the list of indexed attributes then
 203            this node needs a full search */
 204         if (ldb_msg_find_idx(index_list, tree->u.equality.attr, NULL, LTDB_IDXATTR) == -1) {
 205                 return -1;
 206         }
 207
 208         /* the attribute is indexed. Pull the list of DNs that match the
 209            search criterion */
 210         dn = ldb_dn_key(ldb, tree->u.equality.attr, &tree->u.equality.value);
 211         if (!dn) return -1;
 212
 213         msg = talloc(list, struct ldb_message);
 214         if (msg == NULL) {
 215                 return -1;
 216         }
 217
 218         ret = ltdb_search_dn1(module, dn, msg);
 219         talloc_free(dn);
 220         if (ret == 0 || ret == -1) {
 221                 return ret;
 222         }
 223
 224         for (i=0;i<msg->num_elements;i++) {
 225                 struct ldb_message_element *el;
 226
 227                 if (strcmp(msg->elements[i].name, LTDB_IDX) != 0) {
 228                         continue;
 229                 }
 230
 231                 el = &msg->elements[i];
 232
 233                 list->dn = talloc_array(list, char *, el->num_values);
 234                 if (!list->dn) {
 235                         break;
 236                 }
 237
 238                 for (j=0;j<el->num_values;j++) {
 239                         list->dn[list->count] =
 240                                 talloc_strdup(list->dn, (char *)el->values[j].data);
 241                         if (!list->dn[list->count]) {
 242                                 return -1;
 243                         }
 244                         list->count++;
 245                 }
 246         }
 247
 248         talloc_free(msg);
 249
 250         if (list->count > 1) {
 251                 qsort(list->dn, list->count, sizeof(char *), (comparison_fn_t) list_cmp);
 252         }
 253
 254         return 1;
 255 }
 256
 257
 258 static int list_union(struct ldb_context *, struct dn_list *, const struct dn_list *);
 259
 260 /*
 261   return a list of dn's that might match a simple indexed search on
 262   the special objectclass attribute
 263  */
 264 static int ltdb_index_dn_objectclass(struct ldb_module *module,
 265                                      const struct ldb_parse_tree *tree,
 266                                      const struct ldb_message *index_list,
 267                                      struct dn_list *list)
 268 {
 269         struct ldb_context *ldb = module->ldb;
 270         unsigned int i;
 271         int ret;
 272         const char *target = (const char *)tree->u.equality.value.data;
 273         const char **subclasses;
 274
 275         list->count = 0;
 276         list->dn = NULL;
 277
 278         ret = ltdb_index_dn_simple(module, tree, index_list, list);
 279
 280         subclasses = ldb_subclass_list(module->ldb, target);
 281
 282         if (subclasses == NULL) {
 283                 return ret;
 284         }
 285
 286         for (i=0;subclasses[i];i++) {
 287                 struct ldb_parse_tree tree2;
 288                 struct dn_list *list2;
 289                 tree2.operation = LDB_OP_EQUALITY;
 290                 tree2.u.equality.attr = LTDB_OBJECTCLASS;
 291                 if (!tree2.u.equality.attr) {
 292                         return -1;
 293                 }
 294                 tree2.u.equality.value.data =
 295                         (uint8_t *)talloc_strdup(list, subclasses[i]);
 296                 if (tree2.u.equality.value.data == NULL) {
 297                         return -1;
 298                 }
 299                 tree2.u.equality.value.length = strlen(subclasses[i]);
 300                 list2 = talloc(list, struct dn_list);
 301                 if (list2 == NULL) {
 302                         talloc_free(tree2.u.equality.value.data);
 303                         return -1;
 304                 }
 305                 if (ltdb_index_dn_objectclass(module, &tree2,
 306                                               index_list, list2) == 1) {
 307                         if (list->count == 0) {
 308                                 *list = *list2;
 309                                 ret = 1;
 310                         } else {
 311                                 list_union(ldb, list, list2);
 312                                 talloc_free(list2);
 313                         }
 314                 }
 315                 talloc_free(tree2.u.equality.value.data);
 316         }
 317
 318         return ret;
 319 }
 320
 321 /*
 322   return a list of dn's that might match a leaf indexed search
 323  */
 324 static int ltdb_index_dn_leaf(struct ldb_module *module,
 325                               const struct ldb_parse_tree *tree,
 326                               const struct ldb_message *index_list,
 327                               struct dn_list *list)
 328 {
 329         if (ldb_attr_cmp(tree->u.equality.attr, LTDB_OBJECTCLASS) == 0) {
 330                 return ltdb_index_dn_objectclass(module, tree, index_list, list);
 331         }
 332         if (ldb_attr_dn(tree->u.equality.attr) == 0) {
 333                 list->dn = talloc_array(list, char *, 1);
 334                 if (list->dn == NULL) {
 335                         ldb_oom(module->ldb);
 336                         return -1;
 337                 }
 338                 list->dn[0] = talloc_strdup(list, (char *)tree->u.equality.value.data);
 339                 if (list->dn[0] == NULL) {
 340                         ldb_oom(module->ldb);
 341                         return -1;
 342                 }
 343                 list->count = 1;
 344                 return 1;
 345         }
 346         return ltdb_index_dn_simple(module, tree, index_list, list);
 347 }
 348
 349
 350 /*
 351   list intersection
 352   list = list & list2
 353   relies on the lists being sorted
 354 */
 355 static int list_intersect(struct ldb_context *ldb,
 356                           struct dn_list *list, const struct dn_list *list2)
 357 {
 358         struct dn_list *list3;
 359         unsigned int i;
 360
 361         if (list->count == 0 || list2->count == 0) {
 362                 /* 0 & X == 0 */
 363                 return 0;
 364         }
 365
 366         list3 = talloc(ldb, struct dn_list);
 367         if (list3 == NULL) {
 368                 return -1;
 369         }
 370
 371         list3->dn = talloc_array(list3, char *, list->count);
 372         if (!list3->dn) {
 373                 talloc_free(list3);
 374                 return -1;
 375         }
 376         list3->count = 0;
 377
 378         for (i=0;i<list->count;i++) {
 379                 if (ldb_list_find(list->dn[i], list2->dn, list2->count,
 380                               sizeof(char *), (comparison_fn_t)strcmp) != -1) {
 381                         list3->dn[list3->count] = talloc_move(list3->dn, list->dn[i]);
 382                         list3->count++;
 383                 } else {
 384                         talloc_free(list->dn[i]);
 385                 }
 386         }
 387
 388         talloc_free(list->dn);
 389         list->dn = talloc_move(list, list3->dn);
 390         list->count = list3->count;
 391         talloc_free(list3);
 392
 393         return 0;
 394 }
 395
 396
 397 /*
 398   list union
 399   list = list | list2
 400   relies on the lists being sorted
 401 */
 402 static int list_union(struct ldb_context *ldb,
 403                       struct dn_list *list, const struct dn_list *list2)
 404 {
 405         unsigned int i;
 406         char **d;
 407         unsigned int count = list->count;
 408
 409         if (list->count == 0 && list2->count == 0) {
 410                 /* 0 | 0 == 0 */
 411                 return 0;
 412         }
 413
 414         d = talloc_realloc(list, list->dn, char *, list->count + list2->count);
 415         if (!d) {
 416                 return -1;
 417         }
 418         list->dn = d;
 419
 420         for (i=0;i<list2->count;i++) {
 421                 if (ldb_list_find(list2->dn[i], list->dn, count,
 422                               sizeof(char *), (comparison_fn_t)strcmp) == -1) {
 423                         list->dn[list->count] = talloc_strdup(list->dn, list2->dn[i]);
 424                         if (!list->dn[list->count]) {
 425                                 return -1;
 426                         }
 427                         list->count++;
 428                 }
 429         }
 430
 431         if (list->count != count) {
 432                 qsort(list->dn, list->count, sizeof(char *), (comparison_fn_t)list_cmp);
 433         }
 434
 435         return 0;
 436 }
 437
 438 static int ltdb_index_dn(struct ldb_module *module,
 439                          const struct ldb_parse_tree *tree,
 440                          const struct ldb_message *index_list,
 441                          struct dn_list *list);
 442
 443
 444 /*
 445   OR two index results
 446  */
 447 static int ltdb_index_dn_or(struct ldb_module *module,
 448                             const struct ldb_parse_tree *tree,
 449                             const struct ldb_message *index_list,
 450                             struct dn_list *list)
 451 {
 452         struct ldb_context *ldb = module->ldb;
 453         unsigned int i;
 454         int ret;
 455
 456         ret = -1;
 457         list->dn = NULL;
 458         list->count = 0;
 459
 460         for (i=0;i<tree->u.list.num_elements;i++) {
 461                 struct dn_list *list2;
 462                 int v;
 463
 464                 list2 = talloc(module, struct dn_list);
 465                 if (list2 == NULL) {
 466                         return -1;
 467                 }
 468
 469                 v = ltdb_index_dn(module, tree->u.list.elements[i], index_list, list2);
 470
 471                 if (v == 0) {
 472                         /* 0 || X == X */
 473                         if (ret == -1) {
 474                                 ret = 0;
 475                         }
 476                         talloc_free(list2);
 477                         continue;
 478                 }
 479
 480                 if (v == -1) {
 481                         /* 1 || X == 1 */
 482                         talloc_free(list->dn);
 483                         talloc_free(list2);
 484                         return -1;
 485                 }
 486
 487                 if (ret == -1) {
 488                         ret = 1;
 489                         list->dn = talloc_move(list, list2->dn);
 490                         list->count = list2->count;
 491                 } else {
 492                         if (list_union(ldb, list, list2) == -1) {
 493                                 talloc_free(list2);
 494                                 return -1;
 495                         }
 496                         ret = 1;
 497                 }
 498                 talloc_free(list2);
 499         }
 500
 501         if (list->count == 0) {
 502                 return 0;
 503         }
 504
 505         return ret;
 506 }
 507
 508
 509 /*
 510   NOT an index results
 511  */
 512 static int ltdb_index_dn_not(struct ldb_module *module,
 513                              const struct ldb_parse_tree *tree,
 514                              const struct ldb_message *index_list,
 515                              struct dn_list *list)
 516 {
 517         /* the only way to do an indexed not would be if we could
 518            negate the not via another not or if we knew the total
 519            number of database elements so we could know that the
 520            existing expression covered the whole database.
 521
 522            instead, we just give up, and rely on a full index scan
 523            (unless an outer & manages to reduce the list)
 524         */
 525         return -1;
 526 }
 527
 528 /*
 529   AND two index results
 530  */
 531 static int ltdb_index_dn_and(struct ldb_module *module,
 532                              const struct ldb_parse_tree *tree,
 533                              const struct ldb_message *index_list,
 534                              struct dn_list *list)
 535 {
 536         struct ldb_context *ldb = module->ldb;
 537         unsigned int i;
 538         int ret;
 539
 540         ret = -1;
 541         list->dn = NULL;
 542         list->count = 0;
 543
 544         for (i=0;i<tree->u.list.num_elements;i++) {
 545                 struct dn_list *list2;
 546                 int v;
 547
 548                 list2 = talloc(module, struct dn_list);
 549                 if (list2 == NULL) {
 550                         return -1;
 551                 }
 552
 553                 v = ltdb_index_dn(module, tree->u.list.elements[i], index_list, list2);
 554
 555                 if (v == 0) {
 556                         /* 0 && X == 0 */
 557                         talloc_free(list->dn);
 558                         talloc_free(list2);
 559                         return 0;
 560                 }
 561
 562                 if (v == -1) {
 563                         talloc_free(list2);
 564                         continue;
 565                 }
 566
 567                 if (ret == -1) {
 568                         ret = 1;
 569                         talloc_free(list->dn);
 570                         list->dn = talloc_move(list, list2->dn);
 571                         list->count = list2->count;
 572                 } else {
 573                         if (list_intersect(ldb, list, list2) == -1) {
 574                                 talloc_free(list2);
 575                                 return -1;
 576                         }
 577                 }
 578
 579                 talloc_free(list2);
 580
 581                 if (list->count == 0) {
 582                         talloc_free(list->dn);
 583                         return 0;
 584                 }
 585         }
 586
 587         return ret;
 588 }
 589
 590 /*
 591   return a list of dn's that might match a indexed search or
 592   -1 if an error. return 0 for no matches, or 1 for matches
 593  */
 594 static int ltdb_index_dn(struct ldb_module *module,
 595                          const struct ldb_parse_tree *tree,
 596                          const struct ldb_message *index_list,
 597                          struct dn_list *list)
 598 {
 599         int ret = -1;
 600
 601         switch (tree->operation) {
 602         case LDB_OP_AND:
 603                 ret = ltdb_index_dn_and(module, tree, index_list, list);
 604                 break;
 605
 606         case LDB_OP_OR:
 607                 ret = ltdb_index_dn_or(module, tree, index_list, list);
 608                 break;
 609
 610         case LDB_OP_NOT:
 611                 ret = ltdb_index_dn_not(module, tree, index_list, list);
 612                 break;
 613
 614         case LDB_OP_EQUALITY:
 615                 ret = ltdb_index_dn_leaf(module, tree, index_list, list);
 616                 break;
 617
 618         case LDB_OP_SUBSTRING:
 619         case LDB_OP_GREATER:
 620         case LDB_OP_LESS:
 621         case LDB_OP_PRESENT:
 622         case LDB_OP_APPROX:
 623         case LDB_OP_EXTENDED:
 624                 /* we can't index with fancy bitops yet */
 625                 ret = -1;
 626                 break;
 627         }
 628
 629         return ret;
 630 }
 631
 632 /*
 633   filter a candidate dn_list from an indexed search into a set of results
 634   extracting just the given attributes
 635 */
 636 static int ltdb_index_filter(const struct dn_list *dn_list,
 637                              struct ldb_handle *handle)
 638 {
 639         struct ltdb_context *ac = talloc_get_type(handle->private_data, struct ltdb_context);
 640         struct ldb_reply *ares = NULL;
 641         unsigned int i;
 642
 643         for (i = 0; i < dn_list->count; i++) {
 644                 struct ldb_dn *dn;
 645                 int ret;
 646
 647                 ares = talloc_zero(ac, struct ldb_reply);
 648                 if (!ares) {
 649                         handle->status = LDB_ERR_OPERATIONS_ERROR;
 650                         handle->state = LDB_ASYNC_DONE;
 651                         return LDB_ERR_OPERATIONS_ERROR;
 652                 }
 653
 654                 ares->message = ldb_msg_new(ares);
 655                 if (!ares->message) {
 656                         handle->status = LDB_ERR_OPERATIONS_ERROR;
 657                         handle->state = LDB_ASYNC_DONE;
 658                         talloc_free(ares);
 659                         return LDB_ERR_OPERATIONS_ERROR;
 660                 }
 661
 662
 663                 dn = ldb_dn_explode(ares->message, dn_list->dn[i]);
 664                 if (dn == NULL) {
 665                         talloc_free(ares);
 666                         return LDB_ERR_OPERATIONS_ERROR;
 667                 }
 668
 669                 ret = ltdb_search_dn1(ac->module, dn, ares->message);
 670                 talloc_free(dn);
 671                 if (ret == 0) {
 672                         /* the record has disappeared? yes, this can happen */
 673                         talloc_free(ares);
 674                         continue;
 675                 }
 676
 677                 if (ret == -1) {
 678                         /* an internal error */
 679                         talloc_free(ares);
 680                         return LDB_ERR_OPERATIONS_ERROR;
 681                 }
 682
 683                 if (!ldb_match_msg(ac->module->ldb, ares->message, ac->tree, ac->base, ac->scope)) {
 684                         talloc_free(ares);
 685                         continue;
 686                 }
 687
 688                 /* filter the attributes that the user wants */
 689                 ret = ltdb_filter_attrs(ares->message, ac->attrs);
 690
 691                 if (ret == -1) {
 692                         handle->status = LDB_ERR_OPERATIONS_ERROR;
 693                         handle->state = LDB_ASYNC_DONE;
 694                         talloc_free(ares);
 695                         return LDB_ERR_OPERATIONS_ERROR;
 696                 }
 697
 698                 ares->type = LDB_REPLY_ENTRY;
 699                 handle->state = LDB_ASYNC_PENDING;
 700                 handle->status = ac->callback(ac->module->ldb, ac->context, ares);
 701
 702                 if (handle->status != LDB_SUCCESS) {
 703                         handle->state = LDB_ASYNC_DONE;
 704                         return handle->status;
 705                 }
 706         }
 707
 708         return LDB_SUCCESS;
 709 }
 710
 711 /*
 712   search the database with a LDAP-like expression using indexes
 713   returns -1 if an indexed search is not possible, in which
 714   case the caller should call ltdb_search_full()
 715 */
 716 int ltdb_search_indexed(struct ldb_handle *handle)
 717 {
 718         struct ltdb_context *ac = talloc_get_type(handle->private_data, struct ltdb_context);
 719         struct ltdb_private *ltdb = talloc_get_type(ac->module->private_data, struct ltdb_private);
 720         struct dn_list *dn_list;
 721         int ret;
 722
 723         if (ltdb->cache->indexlist->num_elements == 0 &&
 724             ac->scope != LDB_SCOPE_BASE) {
 725                 /* no index list? must do full search */
 726                 return -1;
 727         }
 728
 729         dn_list = talloc(handle, struct dn_list);
 730         if (dn_list == NULL) {
 731                 return -1;
 732         }
 733
 734         if (ac->scope == LDB_SCOPE_BASE) {
 735                 /* with BASE searches only one DN can match */
 736                 dn_list->dn = talloc_array(dn_list, char *, 1);
 737                 if (dn_list->dn == NULL) {
 738                         ldb_oom(ac->module->ldb);
 739                         return -1;
 740                 }
 741                 dn_list->dn[0] = ldb_dn_linearize(dn_list, ac->base);
 742                 if (dn_list->dn[0] == NULL) {
 743                         ldb_oom(ac->module->ldb);
 744                         return -1;
 745                 }
 746                 dn_list->count = 1;
 747                 ret = 1;
 748         } else {
 749                 ret = ltdb_index_dn(ac->module, ac->tree, ltdb->cache->indexlist, dn_list);
 750         }
 751
 752         if (ret == 1) {
 753                 /* we've got a candidate list - now filter by the full tree
 754                    and extract the needed attributes */
 755                 ret = ltdb_index_filter(dn_list, handle);
 756                 handle->status = ret;
 757                 handle->state = LDB_ASYNC_DONE;
 758         }
 759
 760         talloc_free(dn_list);
 761
 762         return ret;
 763 }
 764
 765 /*
 766   add a index element where this is the first indexed DN for this value
 767 */
 768 static int ltdb_index_add1_new(struct ldb_context *ldb,
 769                                struct ldb_message *msg,
 770                                struct ldb_message_element *el,
 771                                const char *dn)
 772 {
 773         struct ldb_message_element *el2;
 774
 775         /* add another entry */
 776         el2 = talloc_realloc(msg, msg->elements,
 777                                struct ldb_message_element, msg->num_elements+1);
 778         if (!el2) {
 779                 return -1;
 780         }
 781
 782         msg->elements = el2;
 783         msg->elements[msg->num_elements].name = talloc_strdup(msg->elements, LTDB_IDX);
 784         if (!msg->elements[msg->num_elements].name) {
 785                 return -1;
 786         }
 787         msg->elements[msg->num_elements].num_values = 0;
 788         msg->elements[msg->num_elements].values = talloc(msg->elements, struct ldb_val);
 789         if (!msg->elements[msg->num_elements].values) {
 790                 return -1;
 791         }
 792         msg->elements[msg->num_elements].values[0].length = strlen(dn);
 793         msg->elements[msg->num_elements].values[0].data = discard_const_p(uint8_t, dn);
 794         msg->elements[msg->num_elements].num_values = 1;
 795         msg->num_elements++;
 796
 797         return 0;
 798 }
 799
 800
 801 /*
 802   add a index element where this is not the first indexed DN for this
 803   value
 804 */
 805 static int ltdb_index_add1_add(struct ldb_context *ldb,
 806                                struct ldb_message *msg,
 807                                struct ldb_message_element *el,
 808                                int idx,
 809                                const char *dn)
 810 {
 811         struct ldb_val *v2;
 812         unsigned int i;
 813
 814         /* for multi-valued attributes we can end up with repeats */
 815         for (i=0;i<msg->elements[idx].num_values;i++) {
 816                 if (strcmp(dn, (char *)msg->elements[idx].values[i].data) == 0) {
 817                         return 0;
 818                 }
 819         }
 820
 821         v2 = talloc_realloc(msg->elements, msg->elements[idx].values,
 822                               struct ldb_val,
 823                               msg->elements[idx].num_values+1);
 824         if (!v2) {
 825                 return -1;
 826         }
 827         msg->elements[idx].values = v2;
 828
 829         msg->elements[idx].values[msg->elements[idx].num_values].length = strlen(dn);
 830         msg->elements[idx].values[msg->elements[idx].num_values].data = discard_const_p(uint8_t, dn);
 831         msg->elements[idx].num_values++;
 832
 833         return 0;
 834 }
 835
 836 /*
 837   add an index entry for one message element
 838 */
 839 static int ltdb_index_add1(struct ldb_module *module, const char *dn,
 840                            struct ldb_message_element *el, int v_idx)
 841 {
 842         struct ldb_context *ldb = module->ldb;
 843         struct ldb_message *msg;
 844         struct ldb_dn *dn_key;
 845         int ret;
 846         unsigned int i;
 847
 848         msg = talloc(module, struct ldb_message);
 849         if (msg == NULL) {
 850                 errno = ENOMEM;
 851                 return -1;
 852         }
 853
 854         dn_key = ldb_dn_key(ldb, el->name, &el->values[v_idx]);
 855         if (!dn_key) {
 856                 talloc_free(msg);
 857                 errno = ENOMEM;
 858                 return -1;
 859         }
 860         talloc_steal(msg, dn_key);
 861
 862         ret = ltdb_search_dn1(module, dn_key, msg);
 863         if (ret == -1) {
 864                 talloc_free(msg);
 865                 return -1;
 866         }
 867
 868         if (ret == 0) {
 869                 msg->dn = dn_key;
 870                 msg->num_elements = 0;
 871                 msg->elements = NULL;
 872         }
 873
 874         for (i=0;i<msg->num_elements;i++) {
 875                 if (strcmp(LTDB_IDX, msg->elements[i].name) == 0) {
 876                         break;
 877                 }
 878         }
 879
 880         if (i == msg->num_elements) {
 881                 ret = ltdb_index_add1_new(ldb, msg, el, dn);
 882         } else {
 883                 ret = ltdb_index_add1_add(ldb, msg, el, i, dn);
 884         }
 885
 886         if (ret == 0) {
 887                 ret = ltdb_store(module, msg, TDB_REPLACE);
 888         }
 889
 890         talloc_free(msg);
 891
 892         return ret;
 893 }
 894
 895 static int ltdb_index_add0(struct ldb_module *module, const char *dn,
 896                            struct ldb_message_element *elements, int num_el)
 897 {
 898         struct ltdb_private *ltdb = module->private_data;
 899         int ret;
 900         unsigned int i, j;
 901
 902         if (dn[0] == '@') {
 903                 return 0;
 904         }
 905
 906         if (ltdb->cache->indexlist->num_elements == 0) {
 907                 /* no indexed fields */
 908                 return 0;
 909         }
 910
 911         for (i = 0; i < num_el; i++) {
 912                 ret = ldb_msg_find_idx(ltdb->cache->indexlist, elements[i].name,
 913                                        NULL, LTDB_IDXATTR);
 914                 if (ret == -1) {
 915                         continue;
 916                 }
 917                 for (j = 0; j < elements[i].num_values; j++) {
 918                         ret = ltdb_index_add1(module, dn, &elements[i], j);
 919                         if (ret == -1) {
 920                                 return -1;
 921                         }
 922                 }
 923         }
 924
 925         return 0;
 926 }
 927
 928 /*
 929   add the index entries for a new record
 930   return -1 on failure
 931 */
 932 int ltdb_index_add(struct ldb_module *module, const struct ldb_message *msg)
 933 {
 934         struct ltdb_private *ltdb = module->private_data;
 935         char *dn;
 936         int ret;
 937
 938         dn = ldb_dn_linearize(ltdb, msg->dn);
 939         if (dn == NULL) {
 940                 return -1;
 941         }
 942
 943         ret = ltdb_index_add0(module, dn, msg->elements, msg->num_elements);
 944
 945         talloc_free(dn);
 946
 947         return ret;
 948 }
 949
 950
 951 /*
 952   delete an index entry for one message element
 953 */
 954 int ltdb_index_del_value(struct ldb_module *module, const char *dn,
 955                          struct ldb_message_element *el, int v_idx)
 956 {
 957         struct ldb_context *ldb = module->ldb;
 958         struct ldb_message *msg;
 959         struct ldb_dn *dn_key;
 960         int ret, i;
 961         unsigned int j;
 962
 963         if (dn[0] == '@') {
 964                 return 0;
 965         }
 966
 967         dn_key = ldb_dn_key(ldb, el->name, &el->values[v_idx]);
 968         if (!dn_key) {
 969                 return -1;
 970         }
 971
 972         msg = talloc(dn_key, struct ldb_message);
 973         if (msg == NULL) {
 974                 talloc_free(dn_key);
 975                 return -1;
 976         }
 977
 978         ret = ltdb_search_dn1(module, dn_key, msg);
 979         if (ret == -1) {
 980                 talloc_free(dn_key);
 981                 return -1;
 982         }
 983
 984         if (ret == 0) {
 985                 /* it wasn't indexed. Did we have an earlier error? If we did then
 986                    its gone now */
 987                 talloc_free(dn_key);
 988                 return 0;
 989         }
 990
 991         i = ldb_msg_find_idx(msg, dn, &j, LTDB_IDX);
 992         if (i == -1) {
 993                 ldb_debug(ldb, LDB_DEBUG_ERROR,
 994                                 "ERROR: dn %s not found in %s\n", dn,
 995                                 ldb_dn_linearize(dn_key, dn_key));
 996                 /* it ain't there. hmmm */
 997                 talloc_free(dn_key);
 998                 return 0;
 999         }
1000
1001         if (j != msg->elements[i].num_values - 1) {
1002                 memmove(&msg->elements[i].values[j],
1003                         &msg->elements[i].values[j+1],
1004                         (msg->elements[i].num_values-(j+1)) *
1005                         sizeof(msg->elements[i].values[0]));
1006         }
1007         msg->elements[i].num_values--;
1008
1009         if (msg->elements[i].num_values == 0) {
1010                 ret = ltdb_delete_noindex(module, dn_key);
1011         } else {
1012                 ret = ltdb_store(module, msg, TDB_REPLACE);
1013         }
1014
1015         talloc_free(dn_key);
1016
1017         return ret;
1018 }
1019
1020 /*
1021   delete the index entries for a record
1022   return -1 on failure
1023 */
1024 int ltdb_index_del(struct ldb_module *module, const struct ldb_message *msg)
1025 {
1026         struct ltdb_private *ltdb = module->private_data;
1027         int ret;
1028         char *dn;
1029         unsigned int i, j;
1030
1031         if (ldb_dn_is_special(msg->dn)) {
1032                 return 0;
1033         }
1034
1035         dn = ldb_dn_linearize(ltdb, msg->dn);
1036         if (dn == NULL) {
1037                 return -1;
1038         }
1039
1040         /* find the list of indexed fields */
1041         if (ltdb->cache->indexlist->num_elements == 0) {
1042                 /* no indexed fields */
1043                 return 0;
1044         }
1045
1046         for (i = 0; i < msg->num_elements; i++) {
1047                 ret = ldb_msg_find_idx(ltdb->cache->indexlist, msg->elements[i].name,
1048                                        NULL, LTDB_IDXATTR);
1049                 if (ret == -1) {
1050                         continue;
1051                 }
1052                 for (j = 0; j < msg->elements[i].num_values; j++) {
1053                         ret = ltdb_index_del_value(module, dn, &msg->elements[i], j);
1054                         if (ret == -1) {
1055                                 talloc_free(dn);
1056                                 return -1;
1057                         }
1058                 }
1059         }
1060
1061         talloc_free(dn);
1062         return 0;
1063 }
1064
1065
1066 /*
1067   traversal function that deletes all @INDEX records
1068 */
1069 static int delete_index(struct tdb_context *tdb, TDB_DATA key, TDB_DATA data, void *state)
1070 {
1071         const char *dn = "DN=" LTDB_INDEX ":";
1072         if (strncmp((char *)key.dptr, dn, strlen(dn)) == 0) {
1073                 return tdb_delete(tdb, key);
1074         }
1075         return 0;
1076 }
1077
1078 /*
1079   traversal function that adds @INDEX records during a re index
1080 */
1081 static int re_index(struct tdb_context *tdb, TDB_DATA key, TDB_DATA data, void *state)
1082 {
1083         struct ldb_module *module = state;
1084         struct ldb_message *msg;
1085         char *dn = NULL;
1086         int ret;
1087         TDB_DATA key2;
1088
1089         if (strncmp((char *)key.dptr, "DN=@", 4) == 0 ||
1090             strncmp((char *)key.dptr, "DN=", 3) != 0) {
1091                 return 0;
1092         }
1093
1094         msg = talloc(module, struct ldb_message);
1095         if (msg == NULL) {
1096                 return -1;
1097         }
1098
1099         ret = ltdb_unpack_data(module, &data, msg);
1100         if (ret != 0) {
1101                 talloc_free(msg);
1102                 return -1;
1103         }
1104
1105         /* check if the DN key has changed, perhaps due to the
1106            case insensitivity of an element changing */
1107         key2 = ltdb_key(module, msg->dn);
1108         if (key2.dptr == NULL) {
1109                 /* probably a corrupt record ... darn */
1110                 ldb_debug(module->ldb, LDB_DEBUG_ERROR, "Invalid DN in re_index: %s\n",
1111                                                         ldb_dn_linearize(msg, msg->dn));
1112                 talloc_free(msg);
1113                 return 0;
1114         }
1115         if (strcmp((char *)key2.dptr, (char *)key.dptr) != 0) {
1116                 tdb_delete(tdb, key);
1117                 tdb_store(tdb, key2, data, 0);
1118         }
1119         talloc_free(key2.dptr);
1120
1121         if (msg->dn == NULL) {
1122                 dn = (char *)key.dptr + 3;
1123         } else {
1124                 dn = ldb_dn_linearize(msg->dn, msg->dn);
1125         }
1126
1127         ret = ltdb_index_add0(module, dn, msg->elements, msg->num_elements);
1128
1129         talloc_free(msg);
1130
1131         return ret;
1132 }
1133
1134 /*
1135   force a complete reindex of the database
1136 */
1137 int ltdb_reindex(struct ldb_module *module)
1138 {
1139         struct ltdb_private *ltdb = module->private_data;
1140         int ret;
1141
1142         if (ltdb_cache_reload(module) != 0) {
1143                 return -1;
1144         }
1145
1146         /* first traverse the database deleting any @INDEX records */
1147         ret = tdb_traverse(ltdb->tdb, delete_index, NULL);
1148         if (ret == -1) {
1149                 return -1;
1150         }
1151
1152         /* now traverse adding any indexes for normal LDB records */
1153         ret = tdb_traverse(ltdb->tdb, re_index, module);
1154         if (ret == -1) {
1155                 return -1;
1156         }
1157
1158         return 0;
1159 }