ITS#7473 Scope-based searches

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

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 );