r20106: Optional ONE Level indexing for ldb_tdb

To activate it you must modify the @INDEXLIST object adding
the attribute @IDXONE: 1

Ldb test included

Simo.
(This used to be commit ea111795f4016916473ccc05d23c6655e6af1207)

diff --git a/source4/lib/ldb/ldb_tdb/ldb_index.c b/source4/lib/ldb/ldb_tdb/ldb_index.c
index 156d9cd1ced3b8042119b48d2c59f6eb9966d013..19385f5504881faa2c67a1ea11ed13519c486c22 100644 (file)
--- a/source4/lib/ldb/ldb_tdb/ldb_index.c
+++ b/source4/lib/ldb/ldb_tdb/ldb_index.c
@@ -155,8 +155,14 @@ static int ldb_msg_find_idx(const struct ldb_message *msg, const char *attr,
        unsigned int i, j;
        for (i=0;i<msg->num_elements;i++) {
                if (ldb_attr_cmp(msg->elements[i].name, key) == 0) {
-                       const struct ldb_message_element *el = 
-                               &msg->elements[i];
+                       const struct ldb_message_element *el = &msg->elements[i];
+
+                       if (attr == NULL) {
+                               /* in this case we are just looking to see if key is present,
+                                  we are not spearching for a specific index */
+                               return 0;
+                       }
+
                        for (j=0;j<el->num_values;j++) {
                                if (ldb_attr_cmp((char *)el->values[j].data, attr) == 0) {
                                        if (v_idx) {
@@ -583,6 +589,103 @@ static int ltdb_index_dn_and(struct ldb_module *module,
        return ret;
 }
 
+/*
+  AND index results and ONE level special index
+ */
+static int ltdb_index_dn_one(struct ldb_module *module, 
+                            struct ldb_dn *parent_dn,
+                            struct dn_list *list)
+{
+       struct ldb_context *ldb = module->ldb;
+       struct dn_list *list2;
+       struct ldb_message *msg;
+       struct ldb_dn *key;
+       struct ldb_val val;
+       unsigned int i, j;
+       int ret;
+       
+       list2 = talloc_zero(module, struct dn_list);
+       if (list2 == NULL) {
+               return -1;
+       }
+
+       /* the attribute is indexed. Pull the list of DNs that match the 
+          search criterion */
+       val.data = (uint8_t *)((intptr_t)ldb_dn_get_casefold(parent_dn));
+       val.length = strlen((char *)val.data);
+       key = ltdb_index_key(ldb, LTDB_IDXONE, &val);
+       if (!key) {
+               talloc_free(list2);
+               return -1;
+       }
+
+       msg = talloc(list2, struct ldb_message);
+       if (msg == NULL) {
+               talloc_free(list2);
+               return -1;
+       }
+
+       ret = ltdb_search_dn1(module, key, msg);
+       talloc_free(key);
+       if (ret == 0 || ret == -1) {
+               return ret;
+       }
+
+       for (i = 0; i < msg->num_elements; i++) {
+               struct ldb_message_element *el;
+
+               if (strcmp(msg->elements[i].name, LTDB_IDX) != 0) {
+                       continue;
+               }
+
+               el = &msg->elements[i];
+
+               list2->dn = talloc_array(list2, char *, el->num_values);
+               if (!list2->dn) {
+                       talloc_free(list2);
+                       return -1;
+               }
+
+               for (j = 0; j < el->num_values; j++) {
+                       list2->dn[list2->count] = talloc_strdup(list2->dn, (char *)el->values[j].data);
+                       if (!list2->dn[list2->count]) {
+                               talloc_free(list2);
+                               return -1;
+                       }
+                       list2->count++;
+               }
+       }
+
+       if (list2->count == 0) {
+               talloc_free(list2);
+               return 0;
+       }
+
+       if (list2->count > 1) {
+               qsort(list2->dn, list2->count, sizeof(char *), (comparison_fn_t) list_cmp);
+       }
+
+       if (list->count > 0) {
+               if (list_intersect(ldb, list, list2) == -1) {
+                       talloc_free(list2);
+                       return -1;
+               }
+
+               if (list->count == 0) {
+                       talloc_free(list->dn);
+                       talloc_free(list2);
+                       return 0;
+               }
+       } else {
+               list->dn = talloc_move(list, &list2->dn);
+               list->count = list2->count;
+       }
+
+       talloc_free(list2);
+
+       return 1;
+}
+
 /*
   return a list of dn's that might match a indexed search or
   -1 if an error. return 0 for no matches, or 1 for matches
@@ -714,15 +817,29 @@ int ltdb_search_indexed(struct ldb_handle *handle)
        struct ltdb_context *ac = talloc_get_type(handle->private_data, struct ltdb_context);
        struct ltdb_private *ltdb = talloc_get_type(ac->module->private_data, struct ltdb_private);
        struct dn_list *dn_list;
-       int ret;
+       int ret, idxattr, idxone;
+
+       idxattr = idxone = 0;
+       ret = ldb_msg_find_idx(ltdb->cache->indexlist, NULL, NULL, LTDB_IDXATTR);
+       if (ret == 0 ) {
+               idxattr = 1;
+       }
 
-       if (ltdb->cache->indexlist->num_elements == 0 && 
-           ac->scope != LDB_SCOPE_BASE) {
-               /* no index list? must do full search */
+       /* We do one level indexing only if requested */
+       ret = ldb_msg_find_idx(ltdb->cache->indexlist, NULL, NULL, LTDB_IDXONE);
+       if (ret == 0 ) {
+               idxone = 1;
+       }
+
+       if ((ac->scope == LDB_SCOPE_ONELEVEL && (idxattr+idxone == 0)) ||
+           (ac->scope == LDB_SCOPE_SUBTREE && idxattr == 0)) {
+               /* no indexs? must do full search */
                return -1;
        }
 
-       dn_list = talloc(handle, struct dn_list);
+       ret = -1;
+
+       dn_list = talloc_zero(handle, struct dn_list);
        if (dn_list == NULL) {
                return -1;
        }
@@ -741,8 +858,19 @@ int ltdb_search_indexed(struct ldb_handle *handle)
                }
                dn_list->count = 1;
                ret = 1;
-       } else {
+       }
+
+       if (ac->scope != LDB_SCOPE_BASE && idxattr == 1) {
                ret = ltdb_index_dn(ac->module, ac->tree, ltdb->cache->indexlist, dn_list);
+
+               if (ret < 0) {
+                       talloc_free(dn_list);
+                       return ret;
+               }
+       }
+
+       if (ac->scope == LDB_SCOPE_ONELEVEL && idxone == 1) {
+               ret = ltdb_index_dn_one(ac->module, ac->base, dn_list);
        }
 
        if (ret == 1) {
@@ -1050,6 +1178,57 @@ int ltdb_index_del(struct ldb_module *module, const struct ldb_message *msg)
        return 0;
 }
 
+/* 
+  handle special index for one level searches
+*/
+int ltdb_index_one(struct ldb_module *module, const struct ldb_message *msg, int add)
+{
+       struct ltdb_private *ltdb = module->private_data;
+       struct ldb_message_element el;
+       struct ldb_val val;
+       struct ldb_dn *pdn;
+       const char *dn;
+       int ret;
+
+       /* We index for ONE Level only if requested */
+       ret = ldb_msg_find_idx(ltdb->cache->indexlist, NULL, NULL, LTDB_IDXONE);
+       if (ret != 0) {
+               return 0;
+       }
+
+       pdn = ldb_dn_get_parent(module, msg->dn);
+       if (pdn == NULL) {
+               return -1;
+       }
+
+       dn = ldb_dn_get_linearized(msg->dn);
+       if (dn == NULL) {
+               talloc_free(pdn);
+               return -1;
+       }
+
+       val.data = (uint8_t *)((intptr_t)ldb_dn_get_casefold(pdn));
+       if (val.data == NULL) {
+               talloc_free(pdn);
+               return -1;
+       }
+
+       val.length = strlen((char *)val.data);
+       el.name = LTDB_IDXONE;
+       el.values = &val;
+       el.num_values = 1;
+
+       if (add) {
+               ret = ltdb_index_add1(module, dn, &el, 0);
+       } else { /* delete */
+               ret = ltdb_index_del_value(module, dn, &el, 0);
+       }
+
+       talloc_free(pdn);
+
+       return ret;
+}
+
 
 /*
   traversal function that deletes all @INDEX records
@@ -1112,7 +1291,14 @@ static int re_index(struct tdb_context *tdb, TDB_DATA key, TDB_DATA data, void *
                dn = ldb_dn_get_linearized(msg->dn);
        }
 
-       ret = ltdb_index_add0(module, dn, msg->elements, msg->num_elements);
+       ret = ltdb_index_one(module, msg, 1);
+       if (ret == 0) {
+               ret = ltdb_index_add0(module, dn, msg->elements, msg->num_elements);
+       } else {
+               ldb_debug(module->ldb, LDB_DEBUG_ERROR,
+                       "Adding special ONE LEVEL index failed (%s)!\n",
+                       ldb_dn_get_linearized(msg->dn));
+       }
 
        talloc_free(msg);
 

diff --git a/source4/lib/ldb/ldb_tdb/ldb_tdb.c b/source4/lib/ldb/ldb_tdb/ldb_tdb.c
index bf90115f5142d6102ebf6578d0dd4849e9682cbf..09ddca503424c6d8398e96d706a13db3f26b1b7f 100644 (file)
--- a/source4/lib/ldb/ldb_tdb/ldb_tdb.c
+++ b/source4/lib/ldb/ldb_tdb/ldb_tdb.c
@@ -276,6 +276,11 @@ static int ltdb_add_internal(struct ldb_module *module, const struct ldb_message
        }
        
        if (ret == LDB_SUCCESS) {
+               ret = ltdb_index_one(module, msg, 1);
+               if (ret != LDB_SUCCESS) {
+                       return LDB_ERR_OPERATIONS_ERROR;
+               }
+
                ret = ltdb_modified(module, msg->dn);
                if (ret != LDB_SUCCESS) {
                        return LDB_ERR_OPERATIONS_ERROR;
@@ -372,6 +377,13 @@ static int ltdb_delete_internal(struct ldb_module *module, struct ldb_dn *dn)
                return LDB_ERR_NO_SUCH_OBJECT;
        }
 
+       /* remove one level attribute */
+       ret = ltdb_index_one(module, msg, 0);
+       if (ret != LDB_SUCCESS) {
+               talloc_free(msg);
+               return LDB_ERR_OPERATIONS_ERROR;
+       }
+
        /* remove any indexed attributes */
        ret = ltdb_index_del(module, msg);
        if (ret != LDB_SUCCESS) {

diff --git a/source4/lib/ldb/ldb_tdb/ldb_tdb.h b/source4/lib/ldb/ldb_tdb/ldb_tdb.h
index 6b556153add12698b6e915b9893cc22c8ec3685a..956f18688b6fe5bbac4dc2c16a418ae08377f856 100644 (file)
--- a/source4/lib/ldb/ldb_tdb/ldb_tdb.h
+++ b/source4/lib/ldb/ldb_tdb/ldb_tdb.h
@@ -60,6 +60,7 @@ struct ltdb_context {
 #define LTDB_INDEXLIST  "@INDEXLIST"
 #define LTDB_IDX        "@IDX"
 #define LTDB_IDXATTR    "@IDXATTR"
+#define LTDB_IDXONE     "@IDXONE"
 #define LTDB_BASEINFO   "@BASEINFO"
 #define LTDB_ATTRIBUTES "@ATTRIBUTES"
 #define LTDB_SUBCLASSES "@SUBCLASSES"
@@ -83,6 +84,7 @@ struct ldb_parse_tree;
 int ltdb_search_indexed(struct ldb_handle *handle);
 int ltdb_index_add(struct ldb_module *module, const struct ldb_message *msg);
 int ltdb_index_del(struct ldb_module *module, const struct ldb_message *msg);
+int ltdb_index_one(struct ldb_module *module, const struct ldb_message *msg, int add);
 int ltdb_reindex(struct ldb_module *module);
 
 /* The following definitions come from lib/ldb/ldb_tdb/ldb_pack.c  */

diff --git a/source4/lib/ldb/tests/test-tdb-features.sh b/source4/lib/ldb/tests/test-tdb-features.sh
index 6f1afdcf3343fa140cb08a7501354328a4d9066a..09a46d833e5dd2ee6f4c0db4aac1a26f9d187f09 100644 (file)
--- a/source4/lib/ldb/tests/test-tdb-features.sh
+++ b/source4/lib/ldb/tests/test-tdb-features.sh
@@ -117,3 +117,54 @@ checkcount 1 '(test=foo)'
 checkcount 0 '(test=FOO)'
 checkcount 1 '(test=f*o*)'
 
+checkone() {
+    count=$1
+    base="$2"
+    expression="$3"
+    n=`bin/ldbsearch -s one -b "$base" "$expression" | grep '^dn' | wc -l`
+    if [ $n != $count ]; then
+       echo "Got $n but expected $count for $expression"
+       $VALGRIND bin/ldbsearch -s one -b "$base" "$expression"
+       exit 1
+    fi
+    echo "OK: $count $expression"
+}
+
+echo "Removing wildcard attribute"
+cat <<EOF | $VALGRIND bin/ldbmodify || exit 1
+dn: @ATTRIBUTES
+changetype: modify
+delete: *
+*: INTEGER
+EOF
+
+echo "Adding one level indexes"
+cat <<EOF | $VALGRIND bin/ldbmodify || exit 1
+dn: @INDEXLIST
+changetype: modify
+add: @IDXONE
+@IDXONE: 1
+EOF
+
+echo "Testing one level indexed search"
+cat <<EOF | $VALGRIND bin/ldbadd || exit 1
+dn: cn=one,cn=t1,cn=TEST
+objectClass: oneclass
+cn: one
+test: one
+EOF
+checkone 1 "cn=t1,cn=TEST" '(test=one)'
+cat <<EOF | $VALGRIND bin/ldbadd || exit 1
+dn: cn=two,cn=t1,cn=TEST
+objectClass: oneclass
+cn: two
+test: one
+
+dn: cn=three,cn=t1,cn=TEST
+objectClass: oneclass
+cn: three
+test: one
+EOF
+checkone 3 "cn=t1,cn=TEST" '(test=one)'
+checkone 1 "cn=t1,cn=TEST" '(cn=two)'
+