Ticket #537 - Improvement of range search

Fix description: The index range search function index_range
_read_ext was written to call idl_fetch_ext to get an idlist
belonging to one key. Then add it to the main idlist as long
as the key satisfiles the range search filter condition.
This patch introduces a new range search function idl_new_
range_fetch to the new idl code, which generates an idlist
in one idl function that eliminates the redundancy such as
generating idlist and cursor per key.

This patch only implements the new idl version.  If idl_new
is not set, the existing code is executed.

Additionally, idl_new_fetch did not abort the read transaction
even if any error occurred in the transaction. Now, it switches
between commit and abort based upon the result.

https://fedorahosted.org/389/ticket/537

Reviewed by Rich (Thank you!!)

ldap/servers/slapd/back-ldbm/back-ldbm.h

@@ -885,4 +885,12 @@ typedef struct _back_search_result_set
 #define TOMBSTONE_INCLUDED 0x1 /* used by find_entry2modify_only_ext and 
                                   entryrdn_index_read */
 
+#define DBT_FREE_PAYLOAD(d) slapi_ch_free(&((d).data))
+/* 
+ * This only works with normalized keys, which should be ok because
+ * at this point both L and R should have already been normalized.
+ */
+#define DBT_EQ(L,R) \
+    ((L)->dsize == (R)->dsize && !memcmp ((L)->dptr, (R)->dptr, (L)->dsize))
+
 #endif /* _back_ldbm_h_ */

ldap/servers/slapd/back-ldbm/filterindex.c

@@ -616,7 +616,7 @@ range_candidates(
                                    low, NULL, 0, &txn, err, allidslimit);
     } else {
         idl = index_range_read_ext(pb, be, type, (char*)indextype_EQUALITY,
-                                   SLAPI_OP_GREATER_OR_EQUAL,
+                                   SLAPI_OP_LESS_OR_EQUAL,
                                    low, high, 1, &txn, err, allidslimit);
     }
 

ldap/servers/slapd/back-ldbm/idl_new.c

@@ -155,7 +155,8 @@ int idl_new_release_private(struct attrinfo *a)
 	return 0;
 }
 
-IDList * idl_new_fetch(
+IDList *
+idl_new_fetch(
     backend *be, 
     DB* db, 
     DBT *inkey, 
@@ -292,7 +293,7 @@ IDList * idl_new_fetch(
         }
 
         LDAPDebug(LDAP_DEBUG_TRACE, "bulk fetch buffer nids=%d\n", count, 0, 0); 
-#if defined(DB_ALLIDS_ON_READ)	
+#if defined(DB_ALLIDS_ON_READ)
         /* enforce the allids read limit */
         if ((NEW_IDL_NO_ALLID != *flag_err) && (NULL != a) &&
              (idl != NULL) && idl_new_exceeds_allidslimit(count, a, allidslimit)) {
@@ -317,13 +318,14 @@ IDList * idl_new_fetch(
         /* we got another ID, add it to our IDL */
         idl_rc = idl_append_extend(&idl, id);
         if (idl_rc) {
-            LDAPDebug(LDAP_DEBUG_ANY, "unable to extend id list (err=%d)\n", idl_rc);
+            LDAPDebug1Arg(LDAP_DEBUG_ANY,
+                          "unable to extend id list (err=%d)\n", idl_rc);
             idl_free(idl); idl = NULL;
             goto error;
         }
 #if defined(DB_ALLIDS_ON_READ)    
         /* enforce the allids read limit */
-        if ((idl != NULL) && idl_new_exceeds_allidslimit(count, a, allidslimit)) {
+        if (idl && idl_new_exceeds_allidslimit(count, a, allidslimit)) {
             idl->b_nids = 1;
             idl->b_ids[0] = ALLID;
             ret = DB_NOTFOUND; /* fool the code below into thinking that we finished the dups */
@@ -365,11 +367,352 @@ error:
             }
         }
     }
-    dblayer_read_txn_commit(be, &s_txn);
+    if (ret) {
+        dblayer_read_txn_abort(be, &s_txn);
+    } else {
+        dblayer_read_txn_commit(be, &s_txn);
+    }
     *flag_err = ret;
     return idl;
 }
 
+/* 
+ * Perform the range search in the idl layer instead of the index layer
+ * to improve the performance.
+ */
+IDList *
+idl_new_range_fetch(
+    backend *be, 
+    DB* db, 
+    DBT *lowerkey, 
+    DBT *upperkey,
+    DB_TXN *txn, 
+    struct attrinfo *ai,
+    int *flag_err,
+    int allidslimit,
+    int sizelimit,
+    time_t stoptime,
+    int lookthrough_limit,
+    int operator
+)
+{
+    int ret = 0;
+    int idl_rc = 0;
+    DBC *cursor = NULL;
+    IDList *idl = NULL;
+    DBT cur_key;
+    DBT data;
+    ID id = 0;
+    size_t count = 0;
+#ifdef DB_USE_BULK_FETCH
+    /* beware that a large buffer on the stack might cause a stack overflow on some platforms */
+    char buffer[BULK_FETCH_BUFFER_SIZE]; 
+    void *ptr;
+    DBT dataret;
+#endif
+    back_txn s_txn;
+    struct ldbminfo *li = (struct ldbminfo *)be->be_database->plg_private;
+    time_t curtime;
+    void *saved_key = NULL;
+
+    if (NEW_IDL_NOOP == *flag_err)
+    {
+        *flag_err = 0;
+        return NULL;
+    }
+
+    dblayer_txn_init(li, &s_txn);
+    if (txn) {
+        dblayer_read_txn_begin(be, txn, &s_txn);
+    }
+
+    /* Make a cursor */
+    ret = db->cursor(db, txn, &cursor, 0);
+    if (0 != ret) {
+        ldbm_nasty(filename,1,ret);
+        cursor = NULL;
+        goto error;
+    }
+    memset(&data, 0, sizeof(data));
+#ifdef DB_USE_BULK_FETCH
+    data.ulen = sizeof(buffer);
+    data.size = sizeof(buffer);
+    data.data = buffer;
+    data.flags = DB_DBT_USERMEM;
+    memset(&dataret, 0, sizeof(dataret));
+#else
+    data.ulen = sizeof(id);
+    data.size = sizeof(id);
+    data.data = &id;
+    data.flags = DB_DBT_USERMEM;
+#endif
+
+    /*
+     * We're not expecting the key to change in value
+     * so we can just use the input key as a buffer.
+     * This avoids memory management of the key.
+     */
+    memset(&cur_key, 0, sizeof(cur_key));
+    cur_key.ulen = lowerkey->size;
+    cur_key.size = lowerkey->size;
+    saved_key = cur_key.data = slapi_ch_malloc(lowerkey->size);
+    memcpy(cur_key.data, lowerkey->data, lowerkey->size);
+    cur_key.flags = DB_DBT_MALLOC;
+
+    /* Position cursor at the first matching key */
+#ifdef DB_USE_BULK_FETCH
+    ret = cursor->c_get(cursor, &cur_key, &data, DB_SET|DB_MULTIPLE);
+#else
+    ret = cursor->c_get(cursor, &cur_key, &data, DB_SET);
+#endif
+    if (0 != ret) {
+        if (DB_NOTFOUND != ret) {
+#ifdef DB_USE_BULK_FETCH
+            if (ret == DB_BUFFER_SMALL) {
+                LDAPDebug(LDAP_DEBUG_ANY, "database index is corrupt; "
+                          "data item for key %s is too large for our buffer "
+                          "(need=%d actual=%d)\n",
+                          cur_key.data, data.size, data.ulen);
+            }
+#endif
+            ldbm_nasty(filename,2,ret);
+        }
+        goto error; /* Not found is OK, return NULL IDL */
+    }
+
+    /* Iterate over the duplicates, amassing them into an IDL */
+#ifdef DB_USE_BULK_FETCH
+    while (cur_key.data &&
+           (upperkey->data ?
+            ((operator == SLAPI_OP_LESS) ?
+             DBTcmp(&cur_key, upperkey, ai->ai_key_cmp_fn) < 0 :
+             DBTcmp(&cur_key, upperkey, ai->ai_key_cmp_fn) <= 0) :
+            PR_TRUE /* e.g., (x > a) */)) {
+        ID lastid = 0;
+
+        DB_MULTIPLE_INIT(ptr, &data);
+
+        /* lookthrough limit & sizelimit check */
+        if (idl) {
+            if ((lookthrough_limit != -1) &&
+                (idl->b_nids > (ID)lookthrough_limit)) {
+                idl_free(idl);
+                idl = idl_allids( be );
+                LDAPDebug0Args(LDAP_DEBUG_TRACE,
+                          "idl_new_range_fetch - lookthrough_limit exceeded\n");
+                *flag_err = LDAP_ADMINLIMIT_EXCEEDED;
+                goto error;
+            }
+            if ((sizelimit > 0) && (idl->b_nids > (ID)sizelimit)) {
+                LDAPDebug0Args(LDAP_DEBUG_TRACE,
+                               "idl_new_range_fetch - sizelimit exceeded\n");
+                *flag_err = LDAP_SIZELIMIT_EXCEEDED;
+                goto error;
+            }
+        }
+        /* timelimit check */
+        if (stoptime > 0) { /* timelimit is set */
+            curtime = current_time();
+            if (curtime >= stoptime) {
+                LDAPDebug0Args(LDAP_DEBUG_TRACE,
+                              "idl_new_range_fetch - timelimit exceeded\n");
+                *flag_err = LDAP_TIMELIMIT_EXCEEDED;
+                goto error;
+            }
+        }
+        while (PR_TRUE) {
+            DB_MULTIPLE_NEXT(ptr, &data, dataret.data, dataret.size);
+            if (dataret.data == NULL) break;
+            if (ptr == NULL) break;
+
+            if (*(int32_t *)ptr < -1) {
+                LDAPDebug1Arg(LDAP_DEBUG_TRACE, "DB_MULTIPLE buffer is corrupt; "
+                              "next offset [%d] is less than zero\n",
+                              *(int32_t *)ptr);
+                /* retry the read */
+                break;
+            }
+            if (dataret.size != sizeof(ID)) {
+                LDAPDebug(LDAP_DEBUG_ANY, "database index is corrupt; "
+                          "key %s has a data item with the wrong size (%d)\n", 
+                          cur_key.data, dataret.size, 0);
+                goto error;
+            }
+            memcpy(&id, dataret.data, sizeof(ID));
+            if (id == lastid) { /* dup */
+                LDAPDebug1Arg(LDAP_DEBUG_TRACE, "Detedted duplicate id "
+                              "%d due to DB_MULTIPLE error - skipping\n",
+                              id);
+                continue; /* get next one */
+            }
+            /* note the last id read to check for dups */
+            lastid = id;
+            /* we got another ID, add it to our IDL */
+            idl_rc = idl_append_extend(&idl, id);
+            if (idl_rc) {
+                LDAPDebug1Arg(LDAP_DEBUG_ANY,
+                              "unable to extend id list (err=%d)\n", idl_rc);
+                idl_free(idl); idl = NULL;
+                goto error;
+            }
+
+            count++;
+        }
+
+        LDAPDebug(LDAP_DEBUG_TRACE, "bulk fetch buffer nids=%d\n", count, 0, 0); 
+#if defined(DB_ALLIDS_ON_READ)
+        /* enforce the allids read limit */
+        if ((NEW_IDL_NO_ALLID != *flag_err) && ai && (idl != NULL) &&
+            idl_new_exceeds_allidslimit(count, ai, allidslimit)) {
+            idl->b_nids = 1;
+            idl->b_ids[0] = ALLID;
+            ret = DB_NOTFOUND; /* fool the code below into thinking that we finished the dups */
+            break;
+        }
+#endif
+        ret = cursor->c_get(cursor, &cur_key, &data, DB_NEXT_DUP|DB_MULTIPLE);
+        if (ret) {
+            if (DBT_EQ(&cur_key, upperkey)) { /* this is the last key */
+                break;
+            }
+            /* First set the cursor (DB_NEXT_NODUP does not take DB_MULTIPLE) */
+            ret = cursor->c_get(cursor, &cur_key, &data, DB_NEXT_NODUP);
+            if (ret) {
+                break;
+            }
+            /* Read the dup data */
+            ret = cursor->c_get(cursor, &cur_key, &data, DB_SET|DB_MULTIPLE);
+            if (saved_key != cur_key.data) {
+                /* key was allocated in c_get */
+                slapi_ch_free(&saved_key);
+                saved_key = cur_key.data;
+            }
+            if (ret) {
+                break;
+            }
+        }
+    }
+#else
+    while (upperkey->data ?
+           ((operator == SLAPI_OP_LESS) ?
+            DBTcmp(&cur_key, upperkey, ai->ai_key_cmp_fn) < 0 :
+            DBTcmp(&cur_key, upperkey, ai->ai_key_cmp_fn) <= 0) :
+           PR_TRUE /* e.g., (x > a) */) {
+        /* lookthrough limit & sizelimit check */
+        if (idl) {
+            if ((lookthrough_limit != -1) &&
+                (idl->b_nids > (ID)lookthrough_limit)) {
+                idl_free(idl);
+                idl = idl_allids( be );
+                LDAPDebug0Args(LDAP_DEBUG_TRACE,
+                          "idl_new_range_fetch - lookthrough_limit exceeded\n");
+                *flag_err = LDAP_ADMINLIMIT_EXCEEDED;
+                goto error;
+            }
+            if ((sizelimit > 0) && (idl->b_nids > (ID)sizelimit)) {
+                LDAPDebug0Args(LDAP_DEBUG_TRACE,
+                               "idl_new_range_fetch - sizelimit exceeded\n");
+                *flag_err = LDAP_SIZELIMIT_EXCEEDED;
+                goto error;
+            }
+        }
+        /* timelimit check */
+        if (stoptime > 0) { /* timelimit is set */
+            curtime = current_time();
+            if (curtime >= stoptime) {
+                LDAPDebug0Args(LDAP_DEBUG_TRACE,
+                              "idl_new_range_fetch - timelimit exceeded\n");
+                *flag_err = LDAP_TIMELIMIT_EXCEEDED;
+                goto error;
+            }
+        }
+        ret = cursor->c_get(cursor,&cur_key,&data,DB_NEXT_DUP);
+        count++;
+        if (ret) {
+            if (DBT_EQ(&cur_key, upperkey)) { /* this is the last key */
+                break;
+            }
+            DBT_FREE_PAYLOAD(cur_key);
+            ret = cursor->c_get(cursor, &cur_key, &data, DB_NEXT_NODUP);
+            if (saved_key != cur_key.data) {
+                /* key was allocated in c_get */
+                slapi_ch_free(&saved_key);
+                saved_key = cur_key.data;
+            }
+            if (ret) {
+                break;
+            }
+        }
+        /* we got another ID, add it to our IDL */
+        idl_rc = idl_append_extend(&idl, id);
+        if (idl_rc) {
+            LDAPDebug1Arg(LDAP_DEBUG_ANY,
+                          "unable to extend id list (err=%d)\n", idl_rc);
+            idl_free(idl); idl = NULL;
+            goto error;
+        }
+#if defined(DB_ALLIDS_ON_READ)    
+        /* enforce the allids read limit */
+        if (idl && idl_new_exceeds_allidslimit(count, ai, allidslimit)) {
+            idl->b_nids = 1;
+            idl->b_ids[0] = ALLID;
+            ret = DB_NOTFOUND; /* fool the code below into thinking that we finished the dups */
+            break;
+        }
+#endif
+    }
+#endif
+
+    if (ret) {
+        if (ret == DB_NOTFOUND) {
+            ret = 0; /* normal case */
+        } else {
+            idl_free(idl); idl = NULL;
+            ldbm_nasty(filename,59,ret);
+            goto error;
+        }
+    }
+
+    /* check for allids value */
+    if (idl && (idl->b_nids == 1) && (idl->b_ids[0] == ALLID)) {
+        idl_free(idl);
+        idl = idl_allids(be);
+        LDAPDebug1Arg(LDAP_DEBUG_TRACE, "idl_new_fetch %s returns allids\n", 
+                      cur_key.data);
+    } else {
+        LDAPDebug2Args(LDAP_DEBUG_TRACE, "idl_new_fetch %s returns nids=%lu\n", 
+                       cur_key.data, (u_long)IDL_NIDS(idl));
+    }
+
+error:
+    DBT_FREE_PAYLOAD(cur_key);
+    /* Close the cursor */
+    if (NULL != cursor) {
+        int ret2 = cursor->c_close(cursor);
+        if (ret2) {
+            ldbm_nasty(filename,3,ret2);
+            if (!ret) {
+                /* if cursor close returns DEADLOCK, we must bubble that up
+                   to the higher layers for retries */
+                ret = ret2;
+            }
+        }
+    }
+    if (ret) {
+        dblayer_read_txn_abort(be, &s_txn);
+    } else {
+        dblayer_read_txn_commit(be, &s_txn);
+    }
+    *flag_err = ret;
+
+    /* sort idl */
+    if (idl && !ALLIDS(idl)) {
+        qsort((void *)&idl->b_ids[0], idl->b_nids,
+              (size_t)sizeof(ID), idl_sort_cmp);
+    }
+    return idl;
+}
+
 int idl_new_insert_key(
     backend *be, 
     DB* db, 
@@ -405,7 +748,7 @@ int idl_new_insert_key(
             if (NULL != disposition) {
                 *disposition = IDL_INSERT_ALLIDS;
             }
-            goto error;	/* allid: don't bother inserting any more */
+            goto error;  /* allid: don't bother inserting any more */
         }
     } else if (DB_NOTFOUND != ret) {
         ldbm_nasty(filename,12,ret);
@@ -506,7 +849,7 @@ int idl_new_delete_key(
     ret = cursor->c_get(cursor,key,&data,DB_GET_BOTH);
     if (0 == ret) {
         if (id == ALLID) {
-            goto error;	/* allid: never delete it */
+            goto error;  /* allid: never delete it */
         }
     } else {
         if (DB_NOTFOUND == ret) {
@@ -557,19 +900,19 @@ static int idl_new_store_allids(backend *be, DB *db, DBT *key, DB_TXN *txn)
     /* Position cursor at the key */
     ret = cursor->c_get(cursor,key,&data,DB_SET);
     if (ret == 0) {
-	/* We found it, so delete all duplicates */
-	ret = cursor->c_del(cursor,0);
-	while (0 == ret) {
-		ret = cursor->c_get(cursor,key,&data,DB_NEXT_DUP);
-		if (0 != ret) {
-			break;
-		}
-		ret = cursor->c_del(cursor,0);
-	}
-	if (0 != ret && DB_NOTFOUND != ret) {
+        /* We found it, so delete all duplicates */
+        ret = cursor->c_del(cursor,0);
+        while (0 == ret) {
+            ret = cursor->c_get(cursor,key,&data,DB_NEXT_DUP);
+            if (0 != ret) {
+                break;
+            }
+            ret = cursor->c_del(cursor,0);
+        }
+        if (0 != ret && DB_NOTFOUND != ret) {
             ldbm_nasty(filename,54,ret);
             goto error; 
-	} else {
+        } else {
             ret = 0;
         }
     } else {
@@ -602,9 +945,9 @@ error:
     }
     return ret;
 #ifdef KRAZY_K0DE
-	/* If this function is called in "no-allids" mode, then it's a bug */
-	ldbm_nasty(filename,63,0);
-	return -1;
+    /* If this function is called in "no-allids" mode, then it's a bug */
+    ldbm_nasty(filename,63,0);
+    return -1;
 #endif
 }
 #endif
@@ -674,7 +1017,7 @@ int idl_new_store_block(
         ret = cursor->c_put(cursor, key, &data, DB_NODUPDATA);
         if (0 != ret) {
             if (DB_KEYEXIST == ret) {
-                ret = 0;	/* exist is okay */
+                ret = 0;   /* exist is okay */
             } else {
                 ldbm_nasty(filename,48,ret);
                 goto error;

ldap/servers/slapd/back-ldbm/index.c

@@ -1086,7 +1086,7 @@ index_read_ext(
 
    see also dblayer_bt_compare
 */
-static int
+int
 DBTcmp (DBT* L, DBT* R, value_compare_fn_type cmp_fn)
 {
     struct berval Lv;
@@ -1107,16 +1107,6 @@ DBTcmp (DBT* L, DBT* R, value_compare_fn_type cmp_fn)
     return cmp_fn(&Lv, &Rv);
 }
 
-/* This only works with normalized keys, which
-   should be ok because at this point both L and R
-   should have already been normalized
-*/
-#define DBT_EQ(L,R) ((L)->dsize == (R)->dsize &&\
- ! memcmp ((L)->dptr, (R)->dptr, (L)->dsize))
-
-
-#define DBT_FREE_PAYLOAD(d) if ((d).data) {free((d).data);(d).data=NULL;}
-
 /* Steps to the next key without keeping a cursor open */
 /* Returns the new key value in the DBT */
 static int index_range_next_key(DB *db,DBT *key,DB_TXN *db_txn)
@@ -1222,7 +1212,7 @@ index_range_read_ext(
     DBT    upperkey = {0};
     DBT    cur_key = {0};
     DBT    data = {0} ;
-    IDList *idl= NULL;
+    IDList *idl = NULL;
     char   *prefix = NULL;
     char   *realbuf, *nextrealbuf;
     size_t reallen, nextreallen;
@@ -1230,10 +1220,9 @@ index_range_read_ext(
     ID     i;
     struct attrinfo    *ai = NULL;
     int lookthrough_limit = -1; /* default no limit */
-    int retry_count = 0;
     int is_and = 0;
     int sizelimit = 0;
-    time_t curtime, stoptime, optime;
+    time_t curtime, stoptime = 0;
     int timelimit = -1;
     back_search_result_set *sr = NULL;
     int isroot = 0;
@@ -1254,19 +1243,20 @@ index_range_read_ext(
 
     plen = strlen(prefix);
     slapi_pblock_get(pb, SLAPI_SEARCH_IS_AND, &is_and);
-    if (!is_and)
-    {
+    if (!is_and) {
         slapi_pblock_get(pb, SLAPI_SEARCH_SIZELIMIT, &sizelimit);
     }
-    slapi_pblock_get( pb, SLAPI_OPINITIATED_TIME, &optime );
     slapi_pblock_get(pb, SLAPI_SEARCH_TIMELIMIT, &timelimit);
-    stoptime = optime + timelimit;
+    if (timelimit != -1) {
+        time_t optime;
+        slapi_pblock_get(pb, SLAPI_OPINITIATED_TIME, &optime);
+        stoptime = optime + timelimit;
+    }
 
     /*
      * Determine the lookthrough_limit from the PBlock.
      * No limit if there is no search result set and the requestor is root.
      */
-
     slapi_pblock_get( pb, SLAPI_SEARCH_RESULT_SET, &sr );
     if (sr != NULL) {
         /* the normal case */
@@ -1279,8 +1269,8 @@ index_range_read_ext(
         }
     }
 
-    LDAPDebug(LDAP_DEBUG_TRACE, "index_range_read lookthrough_limit=%d\n",
-              lookthrough_limit, 0, 0);
+    LDAPDebug1Arg(LDAP_DEBUG_TRACE, "index_range_read lookthrough_limit=%d\n",
+                  lookthrough_limit);
 
     switch( operator ) {
       case SLAPI_OP_LESS:
@@ -1323,7 +1313,7 @@ index_range_read_ext(
     /* get a cursor so we can walk over the table */
     *err = db->cursor(db,db_txn,&dbc,0);
     if (0 != *err ) {
-                ldbm_nasty(errmsg, 1060, *err);
+        ldbm_nasty(errmsg, 1060, *err);
         LDAPDebug( LDAP_DEBUG_ANY,
             "<= index_range_read(%s,%s) NULL: db->cursor() == %i\n",
             type, prefix, *err );
@@ -1344,7 +1334,7 @@ index_range_read_ext(
         reallen = plen + 1; /* include 0 terminator */
         realbuf = slapi_ch_strdup(prefix);
     }
-    if (range != 1) {
+    if (range != 1) { /* open range search */
         char *tmpbuf = NULL;
         /* this is a search with only one boundary value */
         switch( operator ) {
@@ -1390,12 +1380,12 @@ index_range_read_ext(
                 tmpbuf = slapi_ch_realloc (tmpbuf, cur_key.dsize);
                 memcpy (tmpbuf, cur_key.dptr, cur_key.dsize);
                 DBT_FREE_PAYLOAD(upperkey);
-                upperkey.dptr = tmpbuf;
-                upperkey.dsize = cur_key.dsize;
+                upperkey.dptr = NULL; /* x >= a :no need to check upper bound */
+                upperkey.dsize = 0;
             }
             break;
         }
-    } else {
+    } else { /* closed range search: e.g., (&(x >= a)(x <= b)) */
         /* this is a search with two boundary values (starting and ending) */
         if ( nextval != NULL ) { /* compute a key from nextval */
             const size_t vlen = nextval->bv_len;
@@ -1409,22 +1399,10 @@ index_range_read_ext(
             nextrealbuf = slapi_ch_strdup(prefix);
         }
         /* set up the starting and ending keys for search */ 
-        switch( operator ) {
-          case SLAPI_OP_LESS:
-          case SLAPI_OP_LESS_OR_EQUAL:
-            lowerkey.dptr = nextrealbuf;
-            lowerkey.dsize = nextreallen;
-            upperkey.dptr = realbuf;
-            upperkey.dsize = reallen;
-            break;
-          case SLAPI_OP_GREATER_OR_EQUAL:
-          case SLAPI_OP_GREATER:
-            lowerkey.dptr = realbuf;
-            lowerkey.dsize = reallen;
-            upperkey.dptr = nextrealbuf;
-            upperkey.dsize = nextreallen;
-            break;
-        }
+        lowerkey.dptr = realbuf;
+        lowerkey.dsize = reallen;
+        upperkey.dptr = nextrealbuf;
+        upperkey.dsize = nextreallen;
     }
     /* if (LDAP_DEBUG_FILTER)  {
         char encbuf [BUFSIZ];
@@ -1472,140 +1450,159 @@ index_range_read_ext(
     if (operator == SLAPI_OP_GREATER) {
         *err = index_range_next_key(db,&cur_key,db_txn);
     }
-    while (*err == 0 &&
-           (operator == SLAPI_OP_LESS) ?
-           DBTcmp(&cur_key, &upperkey, ai->ai_key_cmp_fn) < 0 :
-           DBTcmp(&cur_key, &upperkey, ai->ai_key_cmp_fn) <= 0) {
-        /* exit the loop when we either run off the end of the table,
-         * fail to read a key, or read a key that's out of range.
-         */
-        IDList *tmp;
-        /*
-        char encbuf [BUFSIZ];
-        LDAPDebug( LDAP_DEBUG_FILTER, "   cur_key=%s(%li bytes)\n",
-               encoded (&cur_key, encbuf), (long)cur_key.dsize, 0 );
-        */
-        /* Check to see if we've already looked too hard */
-        if (idl != NULL && lookthrough_limit != -1 && idl->b_nids > (ID)lookthrough_limit) {
-            if (NULL != idl) {
-                idl_free(idl);
+    if (idl_get_idl_new()) { /* new idl */
+        idl = idl_new_range_fetch(be, db, &cur_key, &upperkey, db_txn,
+                                  ai, err, allidslimit, sizelimit, stoptime,
+                                  lookthrough_limit, operator);
+    } else { /* old idl */
+        int retry_count = 0;
+        while (*err == 0 &&
+               (upperkey.data &&
+                (operator == SLAPI_OP_LESS) ?
+                DBTcmp(&cur_key, &upperkey, ai->ai_key_cmp_fn) < 0 :
+                DBTcmp(&cur_key, &upperkey, ai->ai_key_cmp_fn) <= 0)) {
+            /* exit the loop when we either run off the end of the table,
+             * fail to read a key, or read a key that's out of range.
+             */
+            IDList *tmp;
+            /*
+            char encbuf [BUFSIZ];
+            LDAPDebug( LDAP_DEBUG_FILTER, "   cur_key=%s(%li bytes)\n",
+                   encoded (&cur_key, encbuf), (long)cur_key.dsize, 0 );
+            */
+            /* lookthrough limit and size limit check */
+            if (idl) {
+                if ((lookthrough_limit != -1) && 
+                    (idl->b_nids > (ID)lookthrough_limit)) {
+                    idl_free(idl);
+                    idl = idl_allids( be );
+                    LDAPDebug0Args(LDAP_DEBUG_TRACE,
+                               "index_range_read lookthrough_limit exceeded\n");
+                    *err = LDAP_ADMINLIMIT_EXCEEDED;
+                    break;
+                }
+                if ((sizelimit > 0) && (idl->b_nids > (ID)sizelimit)) {
+                    LDAPDebug0Args(LDAP_DEBUG_TRACE,
+                                   "index_range_read sizelimit exceeded\n");
+                    *err = LDAP_SIZELIMIT_EXCEEDED;
+                    break;
+                }
             }
-            idl = idl_allids( be );
-            LDAPDebug(LDAP_DEBUG_TRACE, "index_range_read lookthrough_limit exceeded\n",
-                                  0, 0, 0);
-            break;
-        }
-        if (idl != NULL && sizelimit > 0 && idl->b_nids > (ID)sizelimit)
-        {
-            LDAPDebug(LDAP_DEBUG_TRACE, "index_range_read sizelimit exceeded\n",
-                                  0, 0, 0);
-            break;
-        }
-        /* check time limit */
-        curtime = current_time();
-        if ( timelimit != -1 && curtime >= stoptime )
-        {
-            LDAPDebug(LDAP_DEBUG_TRACE, "index_range_read timelimit exceeded\n",
-                                  0, 0, 0);
-            break;
-        }
-
-        /* Check to see if the operation has been abandoned (also happens
-         * when the connection is closed by the client).
-         */
-        if ( slapi_op_abandoned( pb )) {
-            if (NULL != idl) {
-                idl_free(idl);
-                idl = NULL;
+            /* check time limit */
+            if (timelimit != -1) {
+                curtime = current_time();
+                if (curtime >= stoptime) {
+                    LDAPDebug0Args(LDAP_DEBUG_TRACE,
+                                   "index_range_read timelimit exceeded\n");
+                    *err = LDAP_TIMELIMIT_EXCEEDED;
+                    break;
+                }
             }
-            LDAPDebug(LDAP_DEBUG_TRACE,
-                    "index_range_read - operation abandoned\n", 0, 0, 0);
-            break;    /* clean up happens outside the while() loop */
-        }
-
-        /* the cur_key DBT already has the first entry in it when we enter the loop */
-        /* so we process the entry then step to the next one */
-        cur_key.flags = 0;
-        for (retry_count = 0; retry_count < IDL_FETCH_RETRY_COUNT; retry_count++) {
-          *err = NEW_IDL_DEFAULT;
-          tmp = idl_fetch_ext( be, db, &cur_key, NULL, ai, err, allidslimit );
-          if(*err == DB_LOCK_DEADLOCK) {
-            ldbm_nasty("index_range_read retrying transaction", 1090, *err);
+            /* Check to see if the operation has been abandoned (also happens
+             * when the connection is closed by the client).
+             */
+            if ( slapi_op_abandoned( pb )) {
+                if (NULL != idl) {
+                    idl_free(idl);
+                    idl = NULL;
+                }
+                LDAPDebug0Args(LDAP_DEBUG_TRACE,
+                               "index_range_read - operation abandoned\n");
+                break;    /* clean up happens outside the while() loop */
+            }
+    
+            /* the cur_key DBT already has the first entry in it when we enter
+             * the loop, so we process the entry then step to the next one */
+            cur_key.flags = 0;
+            for (retry_count = 0;
+                 retry_count < IDL_FETCH_RETRY_COUNT;
+                 retry_count++) {
+              *err = NEW_IDL_DEFAULT;
+              tmp = idl_fetch_ext(be, db, &cur_key, NULL, ai, err, allidslimit);
+              if(*err == DB_LOCK_DEADLOCK) {
+                ldbm_nasty("index_range_read retrying transaction", 1090, *err);
 #ifdef FIX_TXN_DEADLOCKS
 #error if txn != NULL, have to abort and retry the transaction, not just the fetch
 #endif
-            continue;
-          } else {
-            break;
-          }
-        }
-        if(retry_count == IDL_FETCH_RETRY_COUNT) {
-          ldbm_nasty("index_range_read retry count exceeded",1095,*err);
-        }
-        if (!tmp) {
-            if (slapi_is_loglevel_set(LDAP_DEBUG_TRACE)) {
-                char encbuf[BUFSIZ];
-                LDAPDebug2Args(LDAP_DEBUG_TRACE,
-                               "index_range_read_ext: cur_key=%s(%li bytes) was deleted - skipping\n",
-                               encoded(&cur_key, encbuf), (long)cur_key.dsize);
+                continue;
+              } else {
+                break;
+              }
             }
-        } else {
-            /* idl tmp only contains one id */
-            /* append it at the end here; sort idlist at the end */
-            if (ALLIDS(tmp)) {
-                idl_free(idl);
-                idl = tmp;
+            if(retry_count == IDL_FETCH_RETRY_COUNT) {
+              ldbm_nasty("index_range_read retry count exceeded",1095,*err);
+            }
+            if (!tmp) {
+                if (slapi_is_loglevel_set(LDAP_DEBUG_TRACE)) {
+                    char encbuf[BUFSIZ];
+                    LDAPDebug2Args(LDAP_DEBUG_TRACE,
+                                   "index_range_read_ext: cur_key=%s(%li bytes) was deleted - skipping\n",
+                                   encoded(&cur_key, encbuf), (long)cur_key.dsize);
+                }
             } else {
-                ID id;
-                for (id = idl_firstid(tmp); id != NOID; id = idl_nextid(tmp, id)) {
-                    *err = idl_append_extend(&idl, id);
-                    if (*err) {
-                        ldbm_nasty("index_range_read - failed to generate idlist",
-                                   1097, *err);
+                /* idl tmp only contains one id */
+                /* append it at the end here; sort idlist at the end */
+                if (ALLIDS(tmp)) {
+                    idl_free(idl);
+                    idl = tmp;
+                } else {
+                    ID id;
+                    for (id = idl_firstid(tmp);
+                         id != NOID; id = idl_nextid(tmp, id)) {
+                        *err = idl_append_extend(&idl, id);
+                        if (*err) {
+                            ldbm_nasty("index_range_read - failed to generate idlist",
+                                       1097, *err);
+                        }
                     }
+                    idl_free(tmp);
+                }
+                if (ALLIDS(idl)) {
+                    LDAPDebug0Args(LDAP_DEBUG_TRACE,
+                                   "index_range_read hit an allids value\n");
+                    break;
                 }
-                idl_free(tmp);
             }
-            if (ALLIDS(idl)) {
-                LDAPDebug(LDAP_DEBUG_TRACE, "index_range_read hit an allids value\n",
-                          0, 0, 0);
+            if (DBT_EQ (&cur_key, &upperkey)) { /* this is the last key */
+                break;
+                /* Another c_get would return the same key, with no error. */
+            }
+            data.flags = DB_DBT_MALLOC;
+            cur_key.flags = DB_DBT_MALLOC;
+            *err = index_range_next_key(db,&cur_key,db_txn);
+            /* *err = dbc->c_get(dbc,&cur_key,&data,DB_NEXT); */
+            if (*err == DB_NOTFOUND) {
+                *err = 0;
                 break;
             }
         }
-        if (DBT_EQ (&cur_key, &upperkey)) { /* this is the last key */
-            break;
-            /* Another c_get would return the same key, with no error. */
-        }
-        data.flags = DB_DBT_MALLOC;
-        cur_key.flags = DB_DBT_MALLOC;
-        *err = index_range_next_key(db,&cur_key,db_txn);
-        /* *err = dbc->c_get(dbc,&cur_key,&data,DB_NEXT); */
-        if (*err == DB_NOTFOUND) {
-            *err = 0;
-            break;
+        /* sort idl */
+        if (idl && !ALLIDS(idl)) {
+            qsort((void *)&idl->b_ids[0], idl->b_nids,
+                  (size_t)sizeof(ID), idl_sort_cmp);
         }
     }
-    if (*err) LDAPDebug( LDAP_DEBUG_FILTER, "   dbc->c_get(...DB_NEXT) == %i\n", *err, 0, 0);
+    if (*err) {
+        LDAPDebug1Arg(LDAP_DEBUG_FILTER,
+                      "   dbc->c_get(...DB_NEXT) == %i\n", *err);
+    }
 #ifdef LDAP_DEBUG
-        /* this is for debugging only */
-        if (idl != NULL)
-        {
-            if (ALLIDS(idl)) {
-                LDAPDebug( LDAP_DEBUG_FILTER,
-                        "   idl=ALLIDS\n", 0, 0, 0 );
-            } else {
-                LDAPDebug( LDAP_DEBUG_FILTER,
-                        "   idl->b_nids=%d\n", idl->b_nids, 0, 0 );
-                LDAPDebug( LDAP_DEBUG_FILTER,
-                        "   idl->b_nmax=%d\n", idl->b_nmax, 0, 0 );
-
-                for ( i= 0; i< idl->b_nids; i++)
-                {
-                    LDAPDebug( LDAP_DEBUG_FILTER,
-                                "   idl->b_ids[%d]=%d\n", i, idl->b_ids[i], 0);
-                }
+    /* this is for debugging only */
+    if (idl != NULL) {
+        if (ALLIDS(idl)) {
+            LDAPDebug0Args(LDAP_DEBUG_FILTER, "   idl=ALLIDS\n");
+        } else {
+            LDAPDebug1Arg(LDAP_DEBUG_FILTER,
+                          "   idl->b_nids=%d\n", idl->b_nids);
+            LDAPDebug1Arg(LDAP_DEBUG_FILTER,
+                          "   idl->b_nmax=%d\n", idl->b_nmax);
+
+            for (i = 0; i < idl->b_nids; i++) {
+                LDAPDebug2Args(LDAP_DEBUG_FILTER,
+                               "   idl->b_ids[%d]=%d\n", i, idl->b_ids[i]);
             }
         }
+    }
 #endif
 error:
     index_free_prefix(prefix);
@@ -1613,13 +1610,6 @@ error:
     DBT_FREE_PAYLOAD(upperkey);
 
     dblayer_release_index_file( be, ai, db );
-
-    /* sort idl */
-    if (idl && !ALLIDS(idl)) {
-        qsort((void *)&idl->b_ids[0], idl->b_nids,
-              (size_t)sizeof(ID), idl_sort_cmp);
-    }
-
     LDAPDebug( LDAP_DEBUG_TRACE, "<= index_range_read(%s,%s) %lu candidates\n",
                    type, prefix, (u_long)IDL_NIDS(idl) );
     return( idl );

ldap/servers/slapd/back-ldbm/proto-back-ldbm.h

@@ -283,6 +283,10 @@ int idl_new_compare_dups(
     const DBT *a, 
     const DBT *b
 );
+IDList *idl_new_range_fetch(backend *be, DB* db, DBT *lowerkey, DBT *upperkey,
+                            DB_TXN *txn, struct attrinfo *a, int *flag_err,
+                            int allidslimit, int sizelimit, time_t stoptime,
+                            int lookthrough_limit, int operator);
 
 /*
  * index.c
@@ -300,6 +304,7 @@ IDList* index_read_ext_allids( backend *be, char *type, const char* indextype, c
 IDList* index_range_read( Slapi_PBlock *pb, backend *be, char *type, const char* indextype, int ftype, struct berval* val, struct berval* nextval, int range, back_txn *txn, int *err );
 IDList* index_range_read_ext( Slapi_PBlock *pb, backend *be, char *type, const char* indextype, int ftype, struct berval* val, struct berval* nextval, int range, back_txn *txn, int *err, int allidslimit );
 const char *encode( const struct berval* data, char buf[BUFSIZ] );
+int DBTcmp(DBT* L, DBT* R, value_compare_fn_type cmp_fn);
 
 extern const char* indextype_PRESENCE;
 extern const char* indextype_EQUALITY;