ITS#3647 back-hdb fixes, create unsorted subtree IDLs and sort later.

servers/slapd/back-bdb/dn2id.c

@@ -501,11 +501,17 @@ hdb_dn2id_add(
 
 	DBTzero(&key);
 	DBTzero(&data);
-	key.data = &nid;
 	key.size = sizeof(ID);
 	key.flags = DB_DBT_USERMEM;
 	BDB_ID2DISK( eip->bei_id, &nid );
 
+	/* Delete parent's IDL cache entry */
+	if ( bdb->bi_idl_cache_size ) {
+		key.data = &eip->bei_id;
+		bdb_idl_cache_del( bdb, db, &key );
+	}
+	key.data = &nid;
+
 	/* Need to make dummy root node once. Subsequent attempts
 	 * will fail harmlessly.
 	 */
@@ -518,9 +524,6 @@ hdb_dn2id_add(
 		db->put( db, txn, &key, &data, DB_NODUPDATA );
 	}
 
-	if ( bdb->bi_idl_cache_size ) {
-		bdb_idl_cache_del( bdb, db, &key );
-	}
 	data.data = d;
 	data.size = sizeof(diskNode) + rlen + nrlen;
 	data.flags = DB_DBT_USERMEM;
@@ -558,7 +561,6 @@ hdb_dn2id_delete(
 	DBTzero(&key);
 	key.size = sizeof(ID);
 	key.ulen = key.size;
-	key.data = &nid;
 	key.flags = DB_DBT_USERMEM;
 	BDB_ID2DISK( eip->bei_id, &nid );
 
@@ -568,9 +570,16 @@ hdb_dn2id_delete(
 	data.dlen = data.size;
 	data.flags = DB_DBT_USERMEM | DB_DBT_PARTIAL;
 
+	/* Delete IDL cache entries */
 	if ( bdb->bi_idl_cache_size ) {
+		/* Ours */
+		key.data = &e->e_id;
+		bdb_idl_cache_del( bdb, db, &key );
+		/* Parent's */
+		key.data = &eip->bei_id;
 		bdb_idl_cache_del( bdb, db, &key );
 	}
+	key.data = &nid;
 	rc = db->cursor( db, txn, &cursor, bdb->bi_db_opflags );
 	if ( rc ) return rc;
 
@@ -820,7 +829,7 @@ apply_func(
 	EntryInfo *ei = data;
 	ID *idl = arg;
 
-	bdb_idl_insert( idl, ei->bei_id );
+	bdb_idl_append_one( idl, ei->bei_id );
 	return 0;
 }
 
@@ -915,7 +924,7 @@ hdb_dn2idl_internal(
 					ei.bei_rdn.bv_len = len - sizeof(diskNode)
 						- ei.bei_nrdn.bv_len;
 					ei.bei_rdn.bv_val = d->nrdn + ei.bei_nrdn.bv_len + 1;
-					bdb_idl_insert( cx->tmp, ei.bei_id );
+					bdb_idl_append_one( cx->tmp, ei.bei_id );
 					hdb_cache_load( cx->bdb, &ei, &ei2 );
 				}
 			}
@@ -945,16 +954,14 @@ saveit:
 gotit:
 	if ( !BDB_IDL_IS_ZERO( cx->tmp )) {
 		if ( cx->prefix == DN_SUBTREE_PREFIX ) {
-			if (cx->ei->bei_state & CACHE_ENTRY_NO_GRANDKIDS) {
-				bdb_idl_union( cx->ids, cx->tmp );
-			} else {
+			bdb_idl_append( cx->ids, cx->tmp );
+			if ( !(cx->ei->bei_state & CACHE_ENTRY_NO_GRANDKIDS)) {
 				ID *save, idcurs;
 				EntryInfo *ei = cx->ei;
 				int nokids = 1;
 				save = cx->op->o_tmpalloc( BDB_IDL_SIZEOF( cx->tmp ),
 					cx->op->o_tmpmemctx );
 				BDB_IDL_CPY( save, cx->tmp );
-				bdb_idl_union( cx->ids, cx->tmp );
 
 				idcurs = 0;
 				for ( cx->id = bdb_idl_first( save, &idcurs );
@@ -1007,8 +1014,8 @@ hdb_dn2idl(
 	cx.ei = e->e_id ? BEI(e) : &bdb->bi_cache.c_dntree;
 	cx.bdb = bdb;
 	cx.db = cx.bdb->bi_dn2id->bdi_db;
-	cx.prefix = op->ors_scope == LDAP_SCOPE_ONELEVEL
-		? DN_ONE_PREFIX : DN_SUBTREE_PREFIX;
+	cx.prefix = (op->ors_scope == LDAP_SCOPE_ONELEVEL) ?
+		DN_ONE_PREFIX : DN_SUBTREE_PREFIX;
 	cx.ids = ids;
 	cx.buf = stack;
 	cx.op = op;
@@ -1025,6 +1032,10 @@ hdb_dn2idl(
 
 	DBTzero(&cx.data);
 
-	return hdb_dn2idl_internal(&cx);
+	hdb_dn2idl_internal(&cx);
+	if ( !BDB_IDL_IS_ZERO( ids ))
+		bdb_idl_sort( ids );
+
+	return cx.rc;
 }
 #endif	/* BDB_HIER */

servers/slapd/back-bdb/idl.c

@@ -1207,3 +1207,132 @@ ID bdb_idl_next( ID *ids, ID *cursor )
 	return NOID;
 }
 
+/* Add one ID to an unsorted list. We still maintain a lo/hi reference
+ * for fast range compaction.
+ */
+int bdb_idl_append_one( ID *ids, ID id )
+{
+	unsigned x;
+
+	if (BDB_IDL_IS_RANGE( ids )) {
+		/* if already in range, treat as a dup */
+		if (id >= BDB_IDL_FIRST(ids) && id <= BDB_IDL_LAST(ids))
+			return -1;
+		if (id < BDB_IDL_FIRST(ids))
+			ids[1] = id;
+		else if (id > BDB_IDL_LAST(ids))
+			ids[2] = id;
+		return 0;
+	}
+	if ( ids[0] ) {
+		ID tmp;
+
+		if (id < ids[1]) {
+			tmp = ids[1];
+			ids[1] = id;
+			id = tmp;
+		} else if ( ids[0] > 1 && id > ids[2] ) {
+			tmp = ids[2];
+			ids[2] = id;
+			id = tmp;
+		}
+	}
+	ids[0]++;
+	if ( ids[0] >= BDB_IDL_DB_MAX ) {
+		ids[0] = NOID;
+	} else {
+		ids[ids[0]] = id;
+	}
+	return 0;
+}
+
+/* Append unsorted list b to unsorted list a. Both lists must have their
+ * lowest value in slot 1 and highest value in slot 2.
+ */
+int bdb_idl_append( ID *a, ID *b )
+{
+	ID ida, idb;
+
+	if ( BDB_IDL_IS_ZERO( b ) ) {
+		return 0;
+	}
+
+	if ( BDB_IDL_IS_ZERO( a ) ) {
+		BDB_IDL_CPY( a, b );
+		return 0;
+	}
+
+	if ( BDB_IDL_IS_RANGE( a ) || BDB_IDL_IS_RANGE(b) ||
+		a[0] + b[0] >= BDB_IDL_UM_MAX ) {
+		ida = IDL_MIN( a[1], b[1] );
+		idb = IDL_MAX( a[2], b[2] );
+		a[0] = NOID;
+		a[1] = ida;
+		a[2] = idb;
+		return 0;
+	}
+
+	if ( b[1] < a[1] ) {
+		ida = a[1];
+		a[1] = b[1];
+	} else {
+		ida = b[1];
+	}
+	a[0]++;
+	a[a[0]] = ida;
+
+	if ( b[0] > 1 && b[2] > a[2] ) {
+		ida = a[2];
+		a[2] = b[2];
+	} else {
+		ida = b[2];
+	}
+	a[0]++;
+	a[a[0]] = ida;
+
+	if ( b[0] > 2 ) {
+		int i = b[0] - 2;
+		AC_MEMCPY(a+a[0]+1, b+3, i * sizeof(ID));
+		a[0] += i;
+	}
+	return 0;
+	
+}
+
+/* Sort an IDL using HeapSort */
+static void
+siftDown(ID *ids, int root, int bottom)
+{
+	int child;
+	ID temp;
+
+	temp = ids[root];
+	while ((child=root*2) <= bottom) {
+		if (child < bottom && ids[child] < ids[child + 1])
+			child++;
+
+		if (temp >= ids[child])
+			break;
+		ids[root] = ids[child];
+		root = child;
+	}
+	ids[root] = temp;
+}
+
+void
+bdb_idl_sort( ID *ids )
+{
+	int i;
+	ID temp;
+
+	for (i = ids[0] / 2; i >= 1; i--)
+		siftDown(ids, i, ids[0]);
+
+	for (i = ids[0]; i > 1; i--)
+	{
+		temp = ids[i];
+		ids[i] = ids[1];
+		ids[1] = temp;
+		siftDown(ids, 1, i-1);
+	}
+}

servers/slapd/back-bdb/proto-bdb.h

@@ -249,6 +249,9 @@ bdb_idl_cache_del(
 #define bdb_idl_insert				BDB_SYMBOL(idl_insert)
 #define bdb_idl_intersection		BDB_SYMBOL(idl_intersection)
 #define bdb_idl_union				BDB_SYMBOL(idl_union)
+#define bdb_idl_sort				BDB_SYMBOL(idl_sort)
+#define bdb_idl_append				BDB_SYMBOL(idl_append)
+#define bdb_idl_append_one			BDB_SYMBOL(idl_append_one)
 
 #define bdb_idl_fetch_key			BDB_SYMBOL(idl_fetch_key)
 #define bdb_idl_insert_key			BDB_SYMBOL(idl_insert_key)
@@ -294,6 +297,9 @@ bdb_idl_union(
 ID bdb_idl_first( ID *ids, ID *cursor );
 ID bdb_idl_next( ID *ids, ID *cursor );
 
+void bdb_idl_sort( ID *ids );
+int bdb_idl_append( ID *a, ID *b );
+int bdb_idl_append_one( ID *ids, ID id );
 
 
 /*