ITS#7473 Scope-based searches

Walk subtree if number of subtree entries is smaller than
number of index candidates.

diff --git a/servers/slapd/back-mdb/search.c b/servers/slapd/back-mdb/search.c
index eec106027c25e2ea88b9581670e81915106635c2..3ea6c5d9c53dc5c77e5456c4161dd42075fb275e 100644 (file)
--- a/servers/slapd/back-mdb/search.c
+++ b/servers/slapd/back-mdb/search.c
@@ -316,7 +316,7 @@ int
 mdb_search( Operation *op, SlapReply *rs )
 {
        struct mdb_info *mdb = (struct mdb_info *) op->o_bd->be_private;
-       ID              id, cursor, nsubs;
+       ID              id, cursor, nsubs, ncand;
        ID              lastid = NOID;
        ID              candidates[MDB_IDL_UM_SIZE];
        ID2             *scopes;
@@ -328,7 +328,7 @@ mdb_search( Operation *op, SlapReply *rs )
        int             manageDSAit;
        int             tentries = 0;
        IdScopes        isc;
-       MDB_cursor      *mci;
+       MDB_cursor      *mci, *mcd;
 
        mdb_op_info     opinfo = {{{0}}}, *moi = &opinfo;
        MDB_txn                 *ltid = NULL;
@@ -355,9 +355,16 @@ mdb_search( Operation *op, SlapReply *rs )
                return rs->sr_err;
        }
 
+       rs->sr_err = mdb_cursor_open( ltid, mdb->mi_dn2id, &mcd );
+       if ( rs->sr_err ) {
+               mdb_cursor_close( mci );
+               send_ldap_error( op, rs, LDAP_OTHER, "internal error" );
+               return rs->sr_err;
+       }
+
        scopes = scope_chunk_get( op );
        isc.mt = ltid;
-       isc.mc = NULL;
+       isc.mc = mcd;
        isc.scopes = scopes;
 
        if ( op->ors_deref & LDAP_DEREF_FINDING ) {
@@ -365,7 +372,7 @@ mdb_search( Operation *op, SlapReply *rs )
        }
 dn2entry_retry:
        /* get entry with reader lock */
-       rs->sr_err = mdb_dn2entry( op, ltid, NULL, &op->o_req_ndn, &e, &nsubs, 1 );
+       rs->sr_err = mdb_dn2entry( op, ltid, mcd, &op->o_req_ndn, &e, &nsubs, 1 );
 
        switch(rs->sr_err) {
        case MDB_NOTFOUND:
@@ -528,14 +535,41 @@ dn2entry_retry:
        /* select candidates */
        if ( op->oq_search.rs_scope == LDAP_SCOPE_BASE ) {
                rs->sr_err = base_candidate( op->o_bd, base, candidates );
-
+               ncand = 1;
        } else {
+               if ( op->ors_scope == LDAP_SCOPE_ONELEVEL ) {
+                       size_t nkids;
+                       MDB_val key, data;
+                       key.mv_data = &base->e_id;
+                       key.mv_size = sizeof( ID );
+                       mdb_cursor_get( mcd, &key, &data, MDB_SET );
+                       mdb_cursor_count( mcd, &nkids );
+                       nsubs = nkids - 1;
+               } else if ( !base->e_id ) {
+                       /* we don't maintain nsubs for entryID 0.
+                        * just grab entry count from id2entry stat
+                        */
+                       MDB_stat ms;
+                       mdb_stat( ltid, mdb->mi_id2entry, &ms );
+                       nsubs = ms.ms_entries;
+               }
                MDB_IDL_ZERO( candidates );
                scopes[0].mid = 1;
                scopes[1].mid = base->e_id;
                scopes[1].mval.mv_data = NULL;
                rs->sr_err = search_candidates( op, rs, base,
                        ltid, mci, candidates, scopes );
+               ncand = MDB_IDL_N( candidates );
+               if ( !base->e_id || ncand == NOID ) {
+                       /* grab entry count from id2entry stat
+                        */
+                       MDB_stat ms;
+                       mdb_stat( ltid, mdb->mi_id2entry, &ms );
+                       if ( !base->e_id )
+                               nsubs = ms.ms_entries;
+                       if ( ncand == NOID )
+                               ncand = ms.ms_entries;
+               }
        }
 
        /* start cursor at beginning of candidates.
@@ -553,7 +587,7 @@ dn2entry_retry:
        /* if not root and candidates exceed to-be-checked entries, abort */
        if ( op->ors_limit      /* isroot == FALSE */ &&
                op->ors_limit->lms_s_unchecked != -1 &&
-               MDB_IDL_N(candidates) > (unsigned) op->ors_limit->lms_s_unchecked )
+               ncand > (unsigned) op->ors_limit->lms_s_unchecked )
        {
                rs->sr_err = LDAP_ADMINLIMIT_EXCEEDED;
                send_ldap_result( op, rs );
@@ -564,7 +598,7 @@ dn2entry_retry:
        if ( op->ors_limit == NULL      /* isroot == TRUE */ ||
                !op->ors_limit->lms_s_pr_hide )
        {
-               tentries = MDB_IDL_N(candidates);
+               tentries = ncand;
        }
 
        if ( get_pagedresults( op ) > SLAP_CONTROL_IGNORED ) {
@@ -598,9 +632,21 @@ dn2entry_retry:
                        id = mdb_idl_next( candidates, &cursor );
                goto loop_begin;
        }
+       if ( nsubs < ncand ) {
+               int rc;
+               /* Do scope-based search */
+               isc.id = base->e_id;
+               isc.numrdns = 0;
+               rc = mdb_dn2id_walk( op, &isc );
+               if ( rc )
+                       id = NOID;
+               else
+                       id = isc.id;
+       } else {
+               id = mdb_idl_first( candidates, &cursor );
+       }
 
-       for ( id = mdb_idl_first( candidates, &cursor );
-                 id != NOID ; id = mdb_idl_next( candidates, &cursor ) )
+       while (id != NOID)
        {
                int scopeok;
                MDB_val edata;
@@ -635,6 +681,24 @@ loop_begin:
                }
 
 
+               if ( nsubs < ncand ) {
+                       unsigned i;
+                       /* Is this entry in the candidate list? */
+                       scopeok = 0;
+                       if (MDB_IDL_IS_RANGE( candidates )) {
+                               if ( id >= MDB_IDL_RANGE_FIRST( candidates ) &&
+                                       id <= MDB_IDL_RANGE_LAST( candidates ))
+                                       scopeok = 1;
+                       } else {
+                               i = mdb_idl_search( candidates, id );
+                               if ( candidates[i] == id )
+                                       scopeok = 1;
+                       }
+                       if ( scopeok )
+                               goto scopeok;
+                       goto loop_continue;
+               }
+
                /* Does this candidate actually satisfy the search scope?
                 */
                scopeok = 0;
@@ -680,6 +744,7 @@ loop_begin:
                        goto loop_continue;
                }
 
+scopeok:
                if ( id == base->e_id ) {
                        e = base;
                } else {
@@ -688,6 +753,9 @@ loop_begin:
                        rs->sr_err = mdb_id2edata( op, mci, id, &edata );
                        if ( rs->sr_err == MDB_NOTFOUND ) {
 notfound:
+                               if( nsubs < ncand )
+                                       goto loop_continue;
+
                                if( !MDB_IDL_IS_RANGE(candidates) ) {
                                        /* only complain for non-range IDLs */
                                        Debug( LDAP_DEBUG_TRACE,
@@ -754,8 +822,7 @@ notfound:
                         * deref it when finding, return it.
                         */
                        if ( is_entry_alias(e) &&
-                               ((op->ors_deref & LDAP_DEREF_FINDING) ||
-                                       !bvmatch(&e->e_nname, &op->o_req_ndn)))
+                               ((op->ors_deref & LDAP_DEREF_FINDING) || e != base ))
                        {
                                goto loop_continue;
                        }
@@ -769,8 +836,9 @@ notfound:
                        struct berval pdn, pndn;
                        char *d, *n;
                        int i;
+
                        /* child of base, just append RDNs to base->e_name */
-                       if ( isc.nscope == 1 ) {
+                       if ( nsubs < ncand || isc.nscope == 1 ) {
                                pdn = base->e_name;
                                pndn = base->e_nname;
                        } else {
@@ -786,14 +854,28 @@ notfound:
                        e->e_nname.bv_val = op->o_tmpalloc(e->e_nname.bv_len + 1, op->o_tmpmemctx);
                        d = e->e_name.bv_val;
                        n = e->e_nname.bv_val;
-                       for (i=0; i<isc.numrdns; i++) {
-                               memcpy(d, isc.rdns[i].bv_val, isc.rdns[i].bv_len);
-                               d += isc.rdns[i].bv_len;
-                               *d++ = ',';
-                               memcpy(n, isc.nrdns[i].bv_val, isc.nrdns[i].bv_len);
-                               n += isc.nrdns[i].bv_len;
-                               *n++ = ',';
+                       if (nsubs < ncand) {
+                               /* RDNs are in top-down order */
+                               for (i=isc.numrdns-1; i>=0; i--) {
+                                       memcpy(d, isc.rdns[i].bv_val, isc.rdns[i].bv_len);
+                                       d += isc.rdns[i].bv_len;
+                                       *d++ = ',';
+                                       memcpy(n, isc.nrdns[i].bv_val, isc.nrdns[i].bv_len);
+                                       n += isc.nrdns[i].bv_len;
+                                       *n++ = ',';
+                               }
+                       } else {
+                               /* RDNs are in bottom-up order */
+                               for (i=0; i<isc.numrdns; i++) {
+                                       memcpy(d, isc.rdns[i].bv_val, isc.rdns[i].bv_len);
+                                       d += isc.rdns[i].bv_len;
+                                       *d++ = ',';
+                                       memcpy(n, isc.nrdns[i].bv_val, isc.nrdns[i].bv_len);
+                                       n += isc.nrdns[i].bv_len;
+                                       *n++ = ',';
+                               }
                        }
+
                        if (pdn.bv_len) {
                                memcpy(d, pdn.bv_val, pdn.bv_len+1);
                                memcpy(n, pndn.bv_val, pndn.bv_len+1);
@@ -803,7 +885,7 @@ notfound:
                                e->e_name.bv_len--;
                                e->e_nname.bv_len--;
                        }
-                       if (isc.nscope != 1) {
+                       if (pndn.bv_val != base->e_nname.bv_val) {
                                op->o_tmpfree(pndn.bv_val, op->o_tmpmemctx);
                                op->o_tmpfree(pdn.bv_val, op->o_tmpmemctx);
                        }
@@ -903,6 +985,16 @@ loop_continue:
                        e = NULL;
                        rs->sr_entry = NULL;
                }
+
+               if ( nsubs < ncand ) {
+                       int rc = mdb_dn2id_walk( op, &isc );
+                       if (rc)
+                               id = NOID;
+                       else
+                               id = isc.id;
+               } else {
+                       id = mdb_idl_next( candidates, &cursor );
+               }
        }
 
 nochange:
@@ -919,10 +1011,8 @@ nochange:
        rs->sr_err = LDAP_SUCCESS;
 
 done:
-       if( isc.mc )
-               mdb_cursor_close( isc.mc );
-       if (mci)
-               mdb_cursor_close( mci );
+       mdb_cursor_close( mcd );
+       mdb_cursor_close( mci );
        if ( moi == &opinfo ) {
                mdb_txn_reset( moi->moi_txn );
                LDAP_SLIST_REMOVE( &op->o_extra, &moi->moi_oe, OpExtra, oe_next );