s4-ldb: merged with master

5 files changed, 623 insertions, 523 deletions

diff --git a/source4/lib/ldb/ldb_tdb/ldb_cache.c b/source4/lib/ldb/ldb_tdb/ldb_cache.c
index 2c399686ea..f853023509 100644
--- a/source4/lib/ldb/ldb_tdb/ldb_cache.c
+++ b/source4/lib/ldb/ldb_tdb/ldb_cache.c
@@ -190,8 +190,6 @@ static int ltdb_baseinfo_init(struct ldb_module *module)
 	void *data = ldb_module_get_private(module);
 	struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
 	struct ldb_message *msg;
-	struct ldb_message_element el;
-	struct ldb_val val;
 	int ret;
 	/* the initial sequence number must be different from the one
 	   set in ltdb_cache_free(). Thanks to Jon for pointing this
@@ -202,31 +200,21 @@ static int ltdb_baseinfo_init(struct ldb_module *module)
 
 	ltdb->sequence_number = atof(initial_sequence_number);
 
-	msg = talloc(ltdb, struct ldb_message);
-	if (msg == NULL) {
-		goto failed;
-	}
-
-	msg->num_elements = 1;
-	msg->elements = ⪙
+	msg = ldb_msg_new(ltdb);
 	msg->dn = ldb_dn_new(msg, ldb, LTDB_BASEINFO);
 	if (!msg->dn) {
 		goto failed;
 	}
-	el.name = talloc_strdup(msg, LTDB_SEQUENCE_NUMBER);
-	if (!el.name) {
+
+	if (ldb_msg_add_string(msg, LTDB_SEQUENCE_NUMBER, initial_sequence_number) != 0) {
 		goto failed;
 	}
-	el.values = &val;
-	el.num_values = 1;
-	el.flags = 0;
-	val.data = (uint8_t *)talloc_strdup(msg, initial_sequence_number);
-	if (!val.data) {
+
+	if (ldb_msg_add_string(msg, LTDB_INDEX_VERSION, "1") != 0) {
 		goto failed;
 	}
-	val.length = 1;
-	
-	ret = ltdb_store(module, msg, TDB_INSERT);
+
+	ret = ltdb_store(module, msg, msg, TDB_INSERT);
 
 	talloc_free(msg);
 
@@ -325,6 +313,16 @@ int ltdb_cache_load(struct ldb_module *module)
 	}
 	ltdb->sequence_number = seq;
 
+	/* Determine what index format we are in (updated on reindex) */
+	ltdb->index_version = ldb_msg_find_attr_as_uint64(baseinfo, LTDB_INDEX_VERSION, 0);
+
+	if (ltdb->index_version > 1) {
+		ldb_debug(ldb, LDB_DEBUG_ERROR, 
+			  "Invalid index version %d on database.  This ldb supports only index version 0 and 1",
+			  ltdb->index_version);
+		goto failed;
+	}
+
 	/* Read an interpret database options */
 	options = talloc(ltdb->cache, struct ldb_message);
 	if (options == NULL) goto failed;
@@ -448,13 +446,15 @@ int ltdb_increase_sequence_number(struct ldb_module *module)
 
 	s = ldb_timestring(msg, t);
 	if (s == NULL) {
+		talloc_free(msg);
+		errno = ENOMEM;
 		return LDB_ERR_OPERATIONS_ERROR;
 	}
 
 	val_time.data = (uint8_t *)s;
 	val_time.length = strlen(s);
 
-	ret = ltdb_modify_internal(module, msg);
+	ret = ltdb_modify_internal(module, msg, msg);
 
 	talloc_free(msg);
 
@@ -469,6 +469,50 @@ int ltdb_increase_sequence_number(struct ldb_module *module)
 	return ret;
 }
 
+/*
+  increase the index version number to indicate a database change
+*/
+int ltdb_set_casefold_index(struct ldb_module *module)
+{
+	struct ldb_context *ldb;
+	void *data = ldb_module_get_private(module);
+	struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
+	struct ldb_message *msg;
+	struct ldb_message_element *el;
+
+	int ret;
+		
+	ldb = ldb_module_get_ctx(module);
+
+	msg = ldb_msg_new(ltdb);
+	if (msg == NULL) {
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+	msg->dn = ldb_dn_new(msg, ldb, LTDB_BASEINFO);
+	if (msg->dn == NULL) {
+		talloc_free(msg);
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	if (ldb_msg_add_string(msg, LTDB_INDEX_VERSION, "1") != 0) {
+		talloc_free(msg);
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	el = ldb_msg_find_element(msg, LTDB_INDEX_VERSION);
+	if (!el) {
+		talloc_free(msg);
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+	el->flags = LDB_FLAG_MOD_REPLACE;
+
+	ret = ltdb_modify_internal(module, msg, msg);
+
+	talloc_free(msg);
+
+	return ret;
+}
+
 int ltdb_check_at_attributes_values(const struct ldb_val *value)
 {
 	int i;
diff --git a/source4/lib/ldb/ldb_tdb/ldb_index.c b/source4/lib/ldb/ldb_tdb/ldb_index.c
index 7b8d2c249b..0b96e07a7d 100644
--- a/source4/lib/ldb/ldb_tdb/ldb_index.c
+++ b/source4/lib/ldb/ldb_tdb/ldb_index.c
@@ -33,6 +33,7 @@
 
 #include "ldb_tdb.h"
 #include "dlinklist.h"
+#include "ldb_handlers.h"
 
 /*
   the idxptr code is a bit unusual. The way it works is to replace
@@ -53,13 +54,12 @@
   @INDEX records many times during indexing.
  */
 struct ldb_index_pointer {
-	struct ldb_index_pointer *next, *prev;
-	struct ldb_val value;
+	struct ldb_message_element el;
 };
 
 struct ltdb_idxptr {
 	int num_dns;
-	const char **dn_list;
+	struct TDB_DATA *dn_list;
 	bool repack;
 };
 
@@ -71,57 +71,53 @@ static int ltdb_idxptr_add(struct ldb_module *module, const struct ldb_message *
 	void *data = ldb_module_get_private(module);
 	struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
 	ltdb->idxptr->dn_list = talloc_realloc(ltdb->idxptr, ltdb->idxptr->dn_list, 
-					       const char *, ltdb->idxptr->num_dns+1);
+					       struct TDB_DATA, ltdb->idxptr->num_dns+1);
 	if (ltdb->idxptr->dn_list == NULL) {
 		ltdb->idxptr->num_dns = 0;
 		return LDB_ERR_OPERATIONS_ERROR;
 	}
 	ltdb->idxptr->dn_list[ltdb->idxptr->num_dns] =
-		talloc_strdup(ltdb->idxptr->dn_list, ldb_dn_get_linearized(msg->dn));
-	if (ltdb->idxptr->dn_list[ltdb->idxptr->num_dns] == NULL) {
+		ltdb_key(ltdb->idxptr->dn_list, msg->dn);
+	if (ltdb->idxptr->dn_list[ltdb->idxptr->num_dns].dptr == NULL) {
 		return LDB_ERR_OPERATIONS_ERROR;
 	}
 	ltdb->idxptr->num_dns++;
 	return LDB_SUCCESS;
 }
 
-/* free an idxptr record */
-static int ltdb_free_idxptr(struct ldb_module *module, struct ldb_message_element *el)
+/* return an idxptr record */
+static struct ldb_index_pointer *ltdb_return_idxptr(struct ldb_module *module, struct ldb_message_element *el)
 {
 	struct ldb_val val;
 	struct ldb_index_pointer *ptr;
 
 	if (el->num_values != 1) {
-		return LDB_ERR_OPERATIONS_ERROR;
+		return NULL;
 	}
 
 	val = el->values[0];
 	if (val.length != sizeof(void *)) {
-		return LDB_ERR_OPERATIONS_ERROR;
+		return NULL;
 	}
 
 	ptr = *(struct ldb_index_pointer **)val.data;
 	if (talloc_get_type(ptr, struct ldb_index_pointer) != ptr) {
-		return LDB_ERR_OPERATIONS_ERROR;
-	}
-
-	while (ptr) {
-		struct ldb_index_pointer *tmp = ptr;
-		DLIST_REMOVE(ptr, ptr);
-		talloc_free(tmp);
+		return NULL;
 	}
 
-	return LDB_SUCCESS;
+	return ptr;
 }
 
-
 /* convert from the IDXPTR format to a ldb_message_element format */
-static int ltdb_convert_from_idxptr(struct ldb_module *module, struct ldb_message_element *el)
+static int ltdb_convert_from_idxptr(struct ldb_module *module, struct ldb_message *msg, struct ldb_index_pointer **ptr_out)
 {
 	struct ldb_val val;
-	struct ldb_index_pointer *ptr, *tmp;
-	int i;
-	struct ldb_val *val2;
+	struct ldb_index_pointer *ptr;
+
+	struct ldb_message_element *el = ldb_msg_find_element(msg, LTDB_IDXPTR);
+	if (!el) {
+		return LDB_SUCCESS;
+	}
 
 	if (el->num_values != 1) {
 		return LDB_ERR_OPERATIONS_ERROR;
@@ -137,65 +133,28 @@ static int ltdb_convert_from_idxptr(struct ldb_module *module, struct ldb_messag
 		return LDB_ERR_OPERATIONS_ERROR;
 	}
 
-	/* count the length of the list */
-	for (i=0, tmp = ptr; tmp; tmp=tmp->next) {
-		i++;
-	}
+	*el = ptr->el;
 
-	/* allocate the new values array */
-	val2 = talloc_realloc(NULL, el->values, struct ldb_val, i);
-	if (val2 == NULL) {
-		return LDB_ERR_OPERATIONS_ERROR;
-	}
-	el->values = val2;
-	el->num_values = i;
-
-	/* populate the values array */
-	for (i=0, tmp = ptr; tmp; tmp=tmp->next, i++) {
-		el->values[i].length = tmp->value.length;
-		/* we need to over-allocate here as there are still some places
-		   in ldb that rely on null termination. */
-		el->values[i].data = talloc_size(el->values, tmp->value.length+1);
-		if (el->values[i].data == NULL) {
-			return LDB_ERR_OPERATIONS_ERROR;
-		}
-		memcpy(el->values[i].data, tmp->value.data, tmp->value.length);
-		el->values[i].data[tmp->value.length] = 0;
+	if (ptr_out) {
+		*ptr_out = ptr;
 	}
 
-	/* update the name */
-	el->name = LTDB_IDX;
-
        	return LDB_SUCCESS;
 }
 
 
 /* convert to the IDXPTR format from a ldb_message_element format */
-static int ltdb_convert_to_idxptr(struct ldb_module *module, struct ldb_message_element *el)
+static int ltdb_update_idxptr(struct ldb_module *module, TALLOC_CTX *mem_ctx, 
+			      struct ldb_index_pointer *ptr, 
+			      struct ldb_message_element *el)
 {
-	struct ldb_index_pointer *ptr, *tmp;
-	int i;
 	struct ldb_val *val2;
-	void *data = ldb_module_get_private(module);
-	struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
-
-	ptr = NULL;
-
-	for (i=0;i<el->num_values;i++) {
-		tmp = talloc(ltdb->idxptr, struct ldb_index_pointer);
-		if (tmp == NULL) {
-			return LDB_ERR_OPERATIONS_ERROR;
-		}
-		tmp->value = el->values[i];
-		tmp->value.data = talloc_memdup(tmp, tmp->value.data, tmp->value.length);
-		if (tmp->value.data == NULL) {
-			return LDB_ERR_OPERATIONS_ERROR;
-		}
-		DLIST_ADD(ptr, tmp);
-	}
+	ptr->el = *el;
+	talloc_steal(ptr, el->values);
+	talloc_steal(ptr, el->name);
 
 	/* allocate the new values array */
-	val2 = talloc_realloc(NULL, el->values, struct ldb_val, 1);
+	val2 = talloc_array(mem_ctx, struct ldb_val, 1);
 	if (val2 == NULL) {
 		return LDB_ERR_OPERATIONS_ERROR;
 	}
@@ -211,6 +170,21 @@ static int ltdb_convert_to_idxptr(struct ldb_module *module, struct ldb_message_
        	return LDB_SUCCESS;
 }
 
+/* convert to the IDXPTR format from a ldb_message_element format */
+static int ltdb_convert_to_idxptr(struct ldb_module *module, TALLOC_CTX *mem_ctx, 
+				  struct ldb_message_element *el)
+{
+	struct ldb_index_pointer *ptr;
+	void *data = ldb_module_get_private(module);
+	struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
+
+	ptr = talloc(ltdb->idxptr, struct ldb_index_pointer);
+
+	ltdb_update_idxptr(module, mem_ctx, ptr, el);
+
+       	return LDB_SUCCESS;
+}
+
 
 /* enable the idxptr mode when transactions start */
 int ltdb_index_transaction_start(struct ldb_module *module)
@@ -225,57 +199,72 @@ int ltdb_index_transaction_start(struct ldb_module *module)
   a wrapper around ltdb_search_dn1() which translates pointer based index records
   and maps them into normal ldb message structures
  */
-static int ltdb_search_dn1_index(struct ldb_module *module,
-				struct ldb_dn *dn, struct ldb_message *msg)
+static int ltdb_search_dn1_index_key(struct ldb_module *module,
+				     struct TDB_DATA dn_key, struct ldb_message *msg, 
+				     struct ldb_index_pointer **ptr_out)
 {
-	int ret, i;
-	ret = ltdb_search_dn1(module, dn, msg);
+	int ret;
+	ret = ltdb_search_dn1_key(module, dn_key, msg);
 	if (ret != LDB_SUCCESS) {
 		return ret;
 	}
 
 	/* if this isn't a @INDEX record then don't munge it */
 	if (strncmp(ldb_dn_get_linearized(msg->dn), LTDB_INDEX ":", strlen(LTDB_INDEX) + 1) != 0) {
-		return LDB_ERR_OPERATIONS_ERROR;
+		return LDB_ERR_INVALID_DN_SYNTAX;
 	}
 
-	for (i=0;i<msg->num_elements;i++) {
-		struct ldb_message_element *el = &msg->elements[i];
-		if (strcmp(el->name, LTDB_IDXPTR) == 0) {
-			ret = ltdb_convert_from_idxptr(module, el);
-			if (ret != LDB_SUCCESS) {
-				return ret;
-			}
-		}
+	ret = ltdb_convert_from_idxptr(module, msg, ptr_out);
+	if (ret != LDB_SUCCESS) {
+		return ret;
 	}
 
 	return ret;
 }
 
+/*
+  a wrapper around ltdb_search_dn1() which translates pointer based index records
+  and maps them into normal ldb message structures
+ */
+static int ltdb_search_dn1_index(struct ldb_module *module,
+				 struct ldb_dn *dn, struct ldb_message *msg, 
+				 struct ldb_index_pointer **ptr_out)
+{
+	int ret;
+	TDB_DATA tdb_key = ltdb_key(msg, dn);
+	if (!tdb_key.dptr) {
+		/* Why could we not get a casefolded form on this DN? */
+		return LDB_ERR_INVALID_DN_SYNTAX;
+	}
 
+	ret = ltdb_search_dn1_index_key(module, tdb_key, msg, ptr_out);
+	talloc_free(tdb_key.dptr);
+	return ret;
+}
 
 /*
   fixup the idxptr for one DN
  */
-static int ltdb_idxptr_fix_dn(struct ldb_module *module, const char *strdn)
+static int ltdb_idxptr_fix_dn(struct ldb_module *module, TALLOC_CTX *mem_ctx, 
+			      struct TDB_DATA dn_key)
 {
 	struct ldb_context *ldb;
-	struct ldb_dn *dn;
-	struct ldb_message *msg = ldb_msg_new(module);
+	struct ldb_message *msg = ldb_msg_new(mem_ctx);
+	struct ldb_index_pointer *ptr = NULL;
 	int ret;
 
 	ldb = ldb_module_get_ctx(module);
 
-	dn = ldb_dn_new(msg, ldb, strdn);
-	if (ltdb_search_dn1_index(module, dn, msg) == LDB_SUCCESS) {
-		ret = ltdb_store(module, msg, TDB_REPLACE);
+	if (ltdb_search_dn1_index_key(module, dn_key, msg, &ptr) == LDB_SUCCESS) {
+		ret = ltdb_store(module, msg, msg, TDB_REPLACE);
+		talloc_free(ptr);
 	}
 	talloc_free(msg);
 	return ret;
 }
 
 /* cleanup the idxptr mode when transaction commits */
-int ltdb_index_transaction_commit(struct ldb_module *module)
+int ltdb_index_transaction_prepare_commit(struct ldb_module *module)
 {
 	int i;
 	void *data = ldb_module_get_private(module);
@@ -284,7 +273,8 @@ int ltdb_index_transaction_commit(struct ldb_module *module)
 	/* fix all the DNs that we have modified */
 	if (ltdb->idxptr) {
 		for (i=0;i<ltdb->idxptr->num_dns;i++) {
-			ltdb_idxptr_fix_dn(module, ltdb->idxptr->dn_list[i]);
+			ltdb_idxptr_fix_dn(module, ltdb->idxptr->dn_list,
+					   ltdb->idxptr->dn_list[i]);
 		}
 
 		if (ltdb->idxptr->repack) {
@@ -294,6 +284,7 @@ int ltdb_index_transaction_commit(struct ldb_module *module)
 
 	talloc_free(ltdb->idxptr);
 	ltdb->idxptr = NULL;
+
 	return LDB_SUCCESS;
 }
 
@@ -314,47 +305,55 @@ int ltdb_index_transaction_cancel(struct ldb_module *module)
 
    WARNING: This modifies the msg which is passed in
 */
-int ltdb_store_idxptr(struct ldb_module *module, const struct ldb_message *msg, int flgs)
+static int ltdb_store_idxptr(struct ldb_module *module, TALLOC_CTX *mem_ctx, 
+			     const struct ldb_message *msg, 
+			     struct ldb_message_element *idx_el, int flgs)
 {
 	void *data = ldb_module_get_private(module);
 	struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
 	int ret;
 
 	if (ltdb->idxptr) {
-		int i;
-		struct ldb_message *msg2 = ldb_msg_new(module);
+		struct ldb_message *msg2;
+		struct ldb_message_element *idxptr_el = NULL;
 
-		/* free any old pointer */
+		/* reuse any old pointer */
+		msg2 = ldb_msg_new(mem_ctx);
 		ret = ltdb_search_dn1(module, msg->dn, msg2);
 		if (ret == 0) {
-			for (i=0;i<msg2->num_elements;i++) {
-				struct ldb_message_element *el = &msg2->elements[i];
-				if (strcmp(el->name, LTDB_IDXPTR) == 0) {
-					ret = ltdb_free_idxptr(module, el);
-					if (ret != LDB_SUCCESS) {
-						return ret;
-					}
-				}
-			}
+			idxptr_el = ldb_msg_find_element(msg2, LTDB_IDXPTR);
 		}
-		talloc_free(msg2);
-
-		for (i=0;i<msg->num_elements;i++) {
-			struct ldb_message_element *el = &msg->elements[i];
-			if (strcmp(el->name, LTDB_IDX) == 0) {
-				ret = ltdb_convert_to_idxptr(module, el);
-				if (ret != LDB_SUCCESS) {
-					return ret;
-				}
+		
+		/* If we have an idxptr record already, then reuse it */
+		if (idxptr_el) {
+			struct ldb_index_pointer *ptr = ltdb_return_idxptr(module, idxptr_el); 
+			talloc_free(msg2);
+			if (!ptr) {
+				return LDB_ERR_OPERATIONS_ERROR;
+			}
+			ret = ltdb_update_idxptr(module, msg->elements, ptr, idx_el);
+			if (ret != LDB_SUCCESS) {
+				talloc_free(msg2);
+				return ret;
+			}
+		} else {
+			talloc_free(msg2);
+			ret = ltdb_convert_to_idxptr(module, msg->elements, idx_el);
+			if (ret != LDB_SUCCESS) {
+				return ret;
+			}
+			/* Otherwise, we must add it to the list of
+			 * things to fix up at the end of the
+			 * transaction */
+			ret = ltdb_idxptr_add(module, msg);
+			if (ret != LDB_SUCCESS) {
+				return ret;
 			}
 		}
-
-		if (ltdb_idxptr_add(module, msg) != 0) {
-			return LDB_ERR_OPERATIONS_ERROR;
-		}
+		/* Make sure we still do the ltdb_store */
 	}
 
-	ret = ltdb_store(module, msg, flgs);
+	ret = ltdb_store(module, mem_ctx, msg, flgs);
 	return ret;
 }
 
@@ -412,7 +411,7 @@ static int ldb_list_find(const void *needle,
 
 struct dn_list {
 	unsigned int count;
-	char **dn;
+	struct ldb_val *dn;
 };
 
 /*
@@ -420,6 +419,7 @@ struct dn_list {
   caller frees
 */
 static struct ldb_dn *ltdb_index_key(struct ldb_context *ldb,
+				     TALLOC_CTX *mem_ctx, 
 				     const char *attr, const struct ldb_val *value,
 				     const struct ldb_schema_attribute **ap)
 {
@@ -428,8 +428,12 @@ static struct ldb_dn *ltdb_index_key(struct ldb_context *ldb,
 	const struct ldb_schema_attribute *a;
 	char *attr_folded;
 	int r;
+	TALLOC_CTX *tmp_ctx = talloc_new(mem_ctx);
+	if (!tmp_ctx) {
+		return NULL;
+	}
 
-	attr_folded = ldb_attr_casefold(ldb, attr);
+	attr_folded = ldb_attr_casefold(tmp_ctx, attr);
 	if (!attr_folded) {
 		return NULL;
 	}
@@ -438,7 +442,7 @@ static struct ldb_dn *ltdb_index_key(struct ldb_context *ldb,
 	if (ap) {
 		*ap = a;
 	}
-	r = a->syntax->canonicalise_fn(ldb, ldb, value, &v);
+	r = a->syntax->canonicalise_fn(ldb, tmp_ctx, value, &v);
 	if (r != LDB_SUCCESS) {
 		const char *errstr = ldb_errstring(ldb);
 		/* canonicalisation can be refused. For example, 
@@ -446,23 +450,19 @@ static struct ldb_dn *ltdb_index_key(struct ldb_context *ldb,
 		   if the value contains a wildcard */
 		ldb_asprintf_errstring(ldb, "Failed to create index key for attribute '%s':%s%s%s",
 				       attr, ldb_strerror(r), (errstr?":":""), (errstr?errstr:""));
-		talloc_free(attr_folded);
+		talloc_free(tmp_ctx);
 		return NULL;
 	}
 	if (ldb_should_b64_encode(ldb, &v)) {
 		char *vstr = ldb_base64_encode(ldb, (char *)v.data, v.length);
 		if (!vstr) return NULL;
-		ret = ldb_dn_new_fmt(ldb, ldb, "%s:%s::%s", LTDB_INDEX, attr_folded, vstr);
-		talloc_free(vstr);
+		ret = ldb_dn_new_fmt(tmp_ctx, ldb, "%s:%s::%s", LTDB_INDEX, attr_folded, vstr);
 	} else {
-		ret = ldb_dn_new_fmt(ldb, ldb, "%s:%s:%.*s", LTDB_INDEX, attr_folded, (int)v.length, (char *)v.data);
+		ret = ldb_dn_new_fmt(tmp_ctx, ldb, "%s:%s:%.*s", LTDB_INDEX, attr_folded, (int)v.length, (char *)v.data);
 	}
 
-	if (v.data != value->data) {
-		talloc_free(v.data);
-	}
-	talloc_free(attr_folded);
-
+	talloc_steal(mem_ctx, ret);
+	talloc_free(tmp_ctx);
 	return ret;
 }
 
@@ -470,7 +470,7 @@ static struct ldb_dn *ltdb_index_key(struct ldb_context *ldb,
   see if a attribute value is in the list of indexed attributes
 */
 static int ldb_msg_find_idx(const struct ldb_message *msg, const char *attr,
-			    unsigned int *v_idx, const char *key)
+			    const char *key)
 {
 	unsigned int i, j;
 	for (i=0;i<msg->num_elements;i++) {
@@ -485,10 +485,8 @@ static int ldb_msg_find_idx(const struct ldb_message *msg, const char *attr,
 
 			for (j=0;j<el->num_values;j++) {
 				if (ldb_attr_cmp((char *)el->values[j].data, attr) == 0) {
-					if (v_idx) {
-						*v_idx = j;
-					}
-					return i;
+					/* We found the index we were looking for */
+					return 0;
 				}
 			}
 		}
@@ -496,88 +494,122 @@ static int ldb_msg_find_idx(const struct ldb_message *msg, const char *attr,
 	return -1;
 }
 
+static int tdb_data_cmp(const struct TDB_DATA *s1, const struct TDB_DATA *s2)
+{
+	struct ldb_val l1, l2;
+	l1.data = s1->dptr;
+	l1.length = s1->dsize;
+	l2.data = s2->dptr;
+	l2.length = s2->dsize;
+	return ldb_comparison_binary(NULL, NULL, &l1, &l2);
+}
+
 /* used in sorting dn lists */
-static int list_cmp(const char **s1, const char **s2)
+static int ldb_val_list_cmp(const struct ldb_val *l1, const struct ldb_val *l2)
 {
-	return strcmp(*s1, *s2);
+	return ldb_comparison_binary(NULL, NULL, l1, l2);
 }
 
 /*
   return a list of dn's that might match a simple indexed search or
  */
-static int ltdb_index_dn_simple(struct ldb_module *module,
-				const struct ldb_parse_tree *tree,
-				const struct ldb_message *index_list,
-				struct dn_list *list)
+static int ltdb_index_load(struct ldb_module *module,
+			   const char *attr, const struct ldb_val *value, 
+			   struct dn_list *list)
 {
 	struct ldb_context *ldb;
-	struct ldb_dn *dn;
+	struct ldb_dn *dn_key;
 	int ret;
-	unsigned int i, j;
+	unsigned int j;
 	struct ldb_message *msg;
+	void *data = ldb_module_get_private(module);
+	struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
+	struct ldb_message_element *el;
 
 	ldb = ldb_module_get_ctx(module);
 
 	list->count = 0;
 	list->dn = NULL;
 
-	/* if the attribute isn't in the list of indexed attributes then
-	   this node needs a full search */
-	if (ldb_msg_find_idx(index_list, tree->u.equality.attr, NULL, LTDB_IDXATTR) == -1) {
+	msg = talloc(list, struct ldb_message);
+	if (msg == NULL) {
 		return LDB_ERR_OPERATIONS_ERROR;
 	}
 
 	/* the attribute is indexed. Pull the list of DNs that match the 
 	   search criterion */
-	dn = ltdb_index_key(ldb, tree->u.equality.attr, &tree->u.equality.value, NULL);
-	if (!dn) return LDB_ERR_OPERATIONS_ERROR;
-
-	msg = talloc(list, struct ldb_message);
-	if (msg == NULL) {
+	dn_key = ltdb_index_key(ldb, msg, attr, value, NULL);
+	if (!dn_key) {
+		talloc_free(msg);
 		return LDB_ERR_OPERATIONS_ERROR;
 	}
 
-	ret = ltdb_search_dn1_index(module, dn, msg);
-	talloc_free(dn);
+	ret = ltdb_search_dn1_index(module, dn_key, msg, NULL);
+	talloc_free(dn_key);
 	if (ret != LDB_SUCCESS) {
+		talloc_free(msg);
 		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 = ldb_msg_find_element(msg, LTDB_IDX);
 
-		el = &msg->elements[i];
+	if (!el) {
+		return LDB_SUCCESS;
+	}
 
-		list->dn = talloc_array(list, char *, el->num_values);
-		if (!list->dn) {
+	if (ltdb->index_version > 0) {
+		list->dn = el->values;
+		list->count = el->num_values;
+	}
+		
+	list->dn = talloc_array(list, struct ldb_val, el->num_values);
+	if (!list->dn) {
+		talloc_free(msg);
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+	
+	/* Normalise the index loaded off the disk into the new format */
+	for (j=0;j<el->num_values;j++) {
+		struct ldb_dn *dn = ldb_dn_from_ldb_val(list->dn, ldb, &el->values[j]);
+		if (!dn) {
 			talloc_free(msg);
-			return LDB_ERR_OPERATIONS_ERROR;
+			return LDB_ERR_INVALID_DN_SYNTAX;
 		}
-
-		for (j=0;j<el->num_values;j++) {
-			list->dn[list->count] =
-				talloc_strdup(list->dn, (char *)el->values[j].data);
-			if (!list->dn[list->count]) {
-				talloc_free(msg);
-				return LDB_ERR_OPERATIONS_ERROR;
-			}
-			list->count++;
+		list->dn[j] = ldb_dn_alloc_casefold_as_ldb_val(list->dn, dn);
+		talloc_free(dn);
+		if (!list->dn[j].data) {
+			talloc_free(msg);
+			return LDB_ERR_INVALID_DN_SYNTAX;
 		}
 	}
 
-	talloc_free(msg);
-
+	/* In the old index version, we must sort the index when
+	 * reading from disk.  In index version 1, the list on disk is
+	 * pre-sorted */
 	if (list->count > 1) {
-		qsort(list->dn, list->count, sizeof(char *), (comparison_fn_t) list_cmp);
+		qsort(list->dn, list->count, sizeof(struct ldb_val), (comparison_fn_t) ldb_val_list_cmp);
 	}
 
 	return LDB_SUCCESS;
 }
 
+/*
+  return a list of dn's that might match a simple indexed search or
+ */
+static int ltdb_index_dn_simple(struct ldb_module *module,
+				const struct ldb_parse_tree *tree,
+				const struct ldb_message *index_list,
+				struct dn_list *list)
+{
+	/* if the attribute isn't in the list of indexed attributes then
+	   this node needs a full search */
+	if (ldb_msg_find_idx(index_list, tree->u.equality.attr, LTDB_IDXATTR) == -1) {
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	return ltdb_index_load(module, tree->u.equality.attr, &tree->u.equality.value, list);
+}
+
 
 static int list_union(struct ldb_context *, struct dn_list *, const struct dn_list *);
 
@@ -593,15 +625,21 @@ static int ltdb_index_dn_leaf(struct ldb_module *module,
 	ldb = ldb_module_get_ctx(module);
 
 	if (ldb_attr_dn(tree->u.equality.attr) == 0) {
-		list->dn = talloc_array(list, char *, 1);
+		struct ldb_dn *target_as_dn;
+		list->dn = talloc_array(list, struct ldb_val, 1);
 		if (list->dn == NULL) {
 			ldb_oom(ldb);
 			return LDB_ERR_OPERATIONS_ERROR;
 		}
-		list->dn[0] = talloc_strdup(list->dn, (char *)tree->u.equality.value.data);
-		if (list->dn[0] == NULL) {
-			ldb_oom(ldb);
-			return LDB_ERR_OPERATIONS_ERROR;
+		target_as_dn = ldb_dn_from_ldb_val(list->dn, ldb, &tree->u.equality.value);
+		if (target_as_dn == NULL) {
+			return LDB_ERR_INVALID_DN_SYNTAX;
+		}
+		list->dn[0] = ldb_dn_alloc_casefold_as_ldb_val(list->dn, target_as_dn);
+		talloc_free(target_as_dn);
+
+		if (list->dn[0].data == NULL) {
+			return LDB_ERR_INVALID_DN_SYNTAX;
 		}
 		list->count = 1;
 		return LDB_SUCCESS;
@@ -615,8 +653,7 @@ static int ltdb_index_dn_leaf(struct ldb_module *module,
   list = list & list2
   relies on the lists being sorted
 */
-static int list_intersect(struct ldb_context *ldb,
-			  struct dn_list *list, const struct dn_list *list2)
+static int list_intersect(struct dn_list *list, const struct dn_list *list2)
 {
 	struct dn_list *list3;
 	unsigned int i;
@@ -626,12 +663,12 @@ static int list_intersect(struct ldb_context *ldb,
 		return LDB_ERR_NO_SUCH_OBJECT;
 	}
 
-	list3 = talloc(ldb, struct dn_list);
+	list3 = talloc(list, struct dn_list);
 	if (list3 == NULL) {
 		return LDB_ERR_OPERATIONS_ERROR;
 	}
 
-	list3->dn = talloc_array(list3, char *, list->count);
+	list3->dn = talloc_array(list3, struct ldb_val, list->count);
 	if (!list3->dn) {
 		talloc_free(list3);
 		return LDB_ERR_OPERATIONS_ERROR;
@@ -639,16 +676,13 @@ static int list_intersect(struct ldb_context *ldb,
 	list3->count = 0;
 
 	for (i=0;i<list->count;i++) {
-		if (ldb_list_find(list->dn[i], list2->dn, list2->count,
-			      sizeof(char *), (comparison_fn_t)strcmp) != -1) {
-			list3->dn[list3->count] = talloc_move(list3->dn, &list->dn[i]);
+		if (ldb_list_find(&list->dn[i], list2->dn, list2->count,
+				  sizeof(struct ldb_val), (comparison_fn_t)ldb_val_list_cmp) != -1) {
+			list3->dn[list3->count] = list->dn[i];
 			list3->count++;
-		} else {
-			talloc_free(list->dn[i]);
 		}
 	}
 
-	talloc_free(list->dn);
 	list->dn = talloc_move(list, &list3->dn);
 	list->count = list3->count;
 	talloc_free(list3);
@@ -666,7 +700,7 @@ static int list_union(struct ldb_context *ldb,
 		      struct dn_list *list, const struct dn_list *list2)
 {
 	unsigned int i;
-	char **d;
+	struct ldb_val *d;
 	unsigned int count = list->count;
 
 	if (list->count == 0 && list2->count == 0) {
@@ -674,25 +708,22 @@ static int list_union(struct ldb_context *ldb,
 		return LDB_ERR_NO_SUCH_OBJECT;
 	}
 
-	d = talloc_realloc(list, list->dn, char *, list->count + list2->count);
+	d = talloc_realloc(list, list->dn, struct ldb_val, list->count + list2->count);
 	if (!d) {
 		return LDB_ERR_OPERATIONS_ERROR;
 	}
 	list->dn = d;
 
 	for (i=0;i<list2->count;i++) {
-		if (ldb_list_find(list2->dn[i], list->dn, count,
-			      sizeof(char *), (comparison_fn_t)strcmp) == -1) {
-			list->dn[list->count] = talloc_strdup(list->dn, list2->dn[i]);
-			if (!list->dn[list->count]) {
-				return LDB_ERR_OPERATIONS_ERROR;
-			}
+		if (ldb_list_find(&list2->dn[i], list->dn, count,
+			      sizeof(struct ldb_val), (comparison_fn_t)ldb_val_list_cmp) == -1) {
+			list->dn[list->count] = list2->dn[i];
 			list->count++;
 		}
 	}
 
 	if (list->count != count) {
-		qsort(list->dn, list->count, sizeof(char *), (comparison_fn_t)list_cmp);
+		qsort(list->dn, list->count, sizeof(struct ldb_val), (comparison_fn_t) ldb_val_list_cmp);
 	}
 
 	return LDB_ERR_NO_SUCH_OBJECT;
@@ -726,7 +757,7 @@ static int ltdb_index_dn_or(struct ldb_module *module,
 		struct dn_list *list2;
 		int v;
 
-		list2 = talloc(module, struct dn_list);
+		list2 = talloc(list, struct dn_list);
 		if (list2 == NULL) {
 			return LDB_ERR_OPERATIONS_ERROR;
 		}
@@ -760,7 +791,6 @@ static int ltdb_index_dn_or(struct ldb_module *module,
 			}
 			ret = LDB_SUCCESS;
 		}
-		talloc_free(list2);
 	}
 
 	if (list->count == 0) {
@@ -838,7 +868,7 @@ static int ltdb_index_dn_and(struct ldb_module *module,
 			}
 			if (is_unique != only_unique) continue;
 			
-			list2 = talloc(module, struct dn_list);
+			list2 = talloc(list, struct dn_list);
 			if (list2 == NULL) {
 				return LDB_ERR_OPERATIONS_ERROR;
 			}
@@ -863,14 +893,12 @@ static int ltdb_index_dn_and(struct ldb_module *module,
 				list->dn = talloc_move(list, &list2->dn);
 				list->count = list2->count;
 			} else {
-				if (list_intersect(ldb, list, list2) == -1) {
+				if (list_intersect(list, list2) == -1) {
 					talloc_free(list2);
 					return LDB_ERR_OPERATIONS_ERROR;
 				}
 			}
 			
-			talloc_free(list2);
-			
 			if (list->count == 0) {
 				talloc_free(list->dn);
 				return LDB_ERR_NO_SUCH_OBJECT;
@@ -892,79 +920,31 @@ static int ltdb_index_dn_one(struct ldb_module *module,
 			     struct ldb_dn *parent_dn,
 			     struct dn_list *list)
 {
-	struct ldb_context *ldb;
 	struct dn_list *list2;
-	struct ldb_message *msg;
-	struct ldb_dn *key;
 	struct ldb_val val;
-	unsigned int i, j;
 	int ret;
 
-	ldb = ldb_module_get_ctx(module);
-
-	list2 = talloc_zero(module, struct dn_list);
+	list2 = talloc_zero(list, struct dn_list);
 	if (list2 == NULL) {
 		return LDB_ERR_OPERATIONS_ERROR;
 	}
 
 	/* the attribute is indexed. Pull the list of DNs that match the
 	   search criterion */
-	val.data = (uint8_t *)((uintptr_t)ldb_dn_get_casefold(parent_dn));
-	val.length = strlen((char *)val.data);
-	key = ltdb_index_key(ldb, LTDB_IDXONE, &val, NULL);
-	if (!key) {
-		talloc_free(list2);
-		return LDB_ERR_OPERATIONS_ERROR;
-	}
-
-	msg = talloc(list2, struct ldb_message);
-	if (msg == NULL) {
+	val = ldb_dn_get_casefold_as_ldb_val(parent_dn);
+	if (!val.data) {
 		talloc_free(list2);
-		return LDB_ERR_OPERATIONS_ERROR;
+		return LDB_ERR_INVALID_DN_SYNTAX;
 	}
 
-	ret = ltdb_search_dn1_index(module, key, msg);
-	talloc_free(key);
+	ret = ltdb_index_load(module, LTDB_IDXONE, &val, list2);
 	if (ret != LDB_SUCCESS) {
-		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 LDB_ERR_OPERATIONS_ERROR;
-		}
-
-		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 LDB_ERR_OPERATIONS_ERROR;
-			}
-			list2->count++;
-		}
-	}
-
-	if (list2->count == 0) {
 		talloc_free(list2);
-		return LDB_ERR_NO_SUCH_OBJECT;
-	}
-
-	if (list2->count > 1) {
-		qsort(list2->dn, list2->count, sizeof(char *), (comparison_fn_t) list_cmp);
+		return ret;
 	}
 
 	if (list->count > 0) {
-		if (list_intersect(ldb, list, list2) == -1) {
+		if (list_intersect(list, list2) == -1) {
 			talloc_free(list2);
 			return LDB_ERR_OPERATIONS_ERROR;
 		}
@@ -979,8 +959,6 @@ static int ltdb_index_dn_one(struct ldb_module *module,
 		list->count = list2->count;
 	}
 
-	talloc_free(list2);
-
 	return LDB_SUCCESS;
 }
 
@@ -1041,22 +1019,22 @@ static int ltdb_index_filter(const struct dn_list *dn_list,
 	ldb = ldb_module_get_ctx(ac->module);
 
 	for (i = 0; i < dn_list->count; i++) {
-		struct ldb_dn *dn;
 		int ret;
+		struct TDB_DATA key;
 
 		msg = ldb_msg_new(ac);
 		if (!msg) {
+			ldb_oom(ldb);
 			return LDB_ERR_OPERATIONS_ERROR;
 		}
 
-		dn = ldb_dn_new(msg, ldb, dn_list->dn[i]);
-		if (dn == NULL) {
-			talloc_free(msg);
+		key = ltdb_key_from_casefold_dn(msg, dn_list->dn[i]);
+		if (!key.dptr) {
 			return LDB_ERR_OPERATIONS_ERROR;
 		}
 
-		ret = ltdb_search_dn1(ac->module, dn, msg);
-		talloc_free(dn);
+		ret = ltdb_search_dn1_key(ac->module, key, msg);
+		talloc_free(key.dptr);
 		if (ret == LDB_ERR_NO_SUCH_OBJECT) {
 			/* the record has disappeared? yes, this can happen */
 			talloc_free(msg);
@@ -1111,13 +1089,13 @@ int ltdb_search_indexed(struct ltdb_context *ac, uint32_t *match_count)
 	ldb = ldb_module_get_ctx(ac->module);
 
 	idxattr = idxone = 0;
-	ret = ldb_msg_find_idx(ltdb->cache->indexlist, NULL, NULL, LTDB_IDXATTR);
+	ret = ldb_msg_find_idx(ltdb->cache->indexlist, NULL, LTDB_IDXATTR);
 	if (ret == 0 ) {
 		idxattr = 1;
 	}
 
 	/* We do one level indexing only if requested */
-	ret = ldb_msg_find_idx(ltdb->cache->indexlist, NULL, NULL, LTDB_IDXONE);
+	ret = ldb_msg_find_idx(ltdb->cache->indexlist, NULL, LTDB_IDXONE);
 	if (ret == 0 ) {
 		idxone = 1;
 	}
@@ -1137,15 +1115,14 @@ int ltdb_search_indexed(struct ltdb_context *ac, uint32_t *match_count)
 
 	if (ac->scope == LDB_SCOPE_BASE) {
 		/* with BASE searches only one DN can match */
-		dn_list->dn = talloc_array(dn_list, char *, 1);
+		dn_list->dn = talloc_array(dn_list, struct ldb_val, 1);
 		if (dn_list->dn == NULL) {
 			ldb_oom(ldb);
 			return LDB_ERR_OPERATIONS_ERROR;
 		}
-		dn_list->dn[0] = ldb_dn_alloc_linearized(dn_list, ac->base);
-		if (dn_list->dn[0] == NULL) {
-			ldb_oom(ldb);
-			return LDB_ERR_OPERATIONS_ERROR;
+		dn_list->dn[0] = ldb_dn_alloc_casefold_as_ldb_val(dn_list->dn, ac->base);
+		if (dn_list->dn[0].data == NULL) {
+			return LDB_ERR_INVALID_DN_SYNTAX;
 		}
 		dn_list->count = 1;
 		ret = LDB_SUCCESS;
@@ -1176,33 +1153,18 @@ int ltdb_search_indexed(struct ltdb_context *ac, uint32_t *match_count)
 */
 static int ltdb_index_add1_new(struct ldb_context *ldb,
 			       struct ldb_message *msg,
-			       const char *dn)
+			       struct ldb_val *casefold_dn, 
+			       struct ldb_message_element **el)
 {
-	struct ldb_message_element *el;
-
-	/* add another entry */
-	el = talloc_realloc(msg, msg->elements,
-			       struct ldb_message_element, msg->num_elements+1);
-	if (!el) {
-		return LDB_ERR_OPERATIONS_ERROR;
-	}
-
-	msg->elements = el;
-	msg->elements[msg->num_elements].name = talloc_strdup(msg->elements, LTDB_IDX);
-	if (!msg->elements[msg->num_elements].name) {
-		return LDB_ERR_OPERATIONS_ERROR;
+	int ret = ldb_msg_add_value(msg, LTDB_IDX, casefold_dn, el);
+	if (ret == LDB_SUCCESS) {
+		talloc_steal((*el)->values, casefold_dn->data);
 	}
-	msg->elements[msg->num_elements].num_values = 0;
-	msg->elements[msg->num_elements].values = talloc(msg->elements, struct ldb_val);
-	if (!msg->elements[msg->num_elements].values) {
-		return LDB_ERR_OPERATIONS_ERROR;
+	if (ret != LDB_SUCCESS) {
+		ldb_oom(ldb);
+		return ret;
 	}
-	msg->elements[msg->num_elements].values[0].length = strlen(dn);
-	msg->elements[msg->num_elements].values[0].data = discard_const_p(uint8_t, dn);
-	msg->elements[msg->num_elements].num_values = 1;
-	msg->num_elements++;
-
-	return LDB_SUCCESS;
+	return ret;
 }
 
 
@@ -1212,16 +1174,16 @@ static int ltdb_index_add1_new(struct ldb_context *ldb,
 */
 static int ltdb_index_add1_add(struct ldb_context *ldb,
 			       struct ldb_message *msg,
-			       int idx,
-			       const char *dn,
+			       struct ldb_message_element *el,
+			       struct ldb_val *casefold_dn,
 			       const struct ldb_schema_attribute *a)
 {
 	struct ldb_val *v2;
 	unsigned int i;
 
 	/* for multi-valued attributes we can end up with repeats */
-	for (i=0;i<msg->elements[idx].num_values;i++) {
-		if (strcmp(dn, (char *)msg->elements[idx].values[i].data) == 0) {
+	for (i=0;i<el->num_values;i++) {
+		if (ldb_comparison_binary(NULL, NULL, casefold_dn, &el->values[i]) == 0) {
 			return LDB_SUCCESS;
 		}
 	}
@@ -1230,17 +1192,22 @@ static int ltdb_index_add1_add(struct ldb_context *ldb,
 		return LDB_ERR_ENTRY_ALREADY_EXISTS;
 	}
 
-	v2 = talloc_realloc(msg->elements, msg->elements[idx].values,
+	v2 = talloc_realloc(msg->elements, el->values,
 			      struct ldb_val,
-			      msg->elements[idx].num_values+1);
+			      el->num_values+1);
 	if (!v2) {
+		ldb_oom(ldb);
 		return LDB_ERR_OPERATIONS_ERROR;
 	}
-	msg->elements[idx].values = v2;
+	el->values = v2;
 
-	msg->elements[idx].values[msg->elements[idx].num_values].length = strlen(dn);
-	msg->elements[idx].values[msg->elements[idx].num_values].data = discard_const_p(uint8_t, dn);
-	msg->elements[idx].num_values++;
+	el->values[el->num_values] = *casefold_dn;
+	el->num_values++;
+	talloc_steal(el->values, casefold_dn->data);
+
+	/* In Index version 1, we must have a sorted index list on
+	 * disk.  Harmless for reading with the old index version. */
+	qsort(el->values, el->num_values, sizeof(struct ldb_val), (comparison_fn_t) ldb_val_list_cmp);
 
 	return LDB_SUCCESS;
 }
@@ -1248,32 +1215,34 @@ static int ltdb_index_add1_add(struct ldb_context *ldb,
 /*
   add an index entry for one message element
 */
-static int ltdb_index_add1(struct ldb_module *module, const char *dn,
+static int ltdb_index_add1(struct ldb_module *module, TALLOC_CTX *mem_ctx, 
+			   struct ldb_dn *dn,
 			   struct ldb_message_element *el, int v_idx)
 {
 	struct ldb_context *ldb;
 	struct ldb_message *msg;
 	struct ldb_dn *dn_key;
 	int ret;
-	unsigned int i;
 	const struct ldb_schema_attribute *a;
+	struct ldb_val casefold_dn;
 
 	ldb = ldb_module_get_ctx(module);
 
-	msg = talloc(module, struct ldb_message);
+	msg = talloc(mem_ctx, struct ldb_message);
 	if (msg == NULL) {
-		errno = ENOMEM;
+		ldb_oom(ldb);
 		return LDB_ERR_OPERATIONS_ERROR;
 	}
 
-	dn_key = ltdb_index_key(ldb, el->name, &el->values[v_idx], &a);
+	dn_key = ltdb_index_key(ldb, msg, el->name, &el->values[v_idx], &a);
 	if (!dn_key) {
 		talloc_free(msg);
 		return LDB_ERR_OPERATIONS_ERROR;
 	}
 	talloc_steal(msg, dn_key);
 
-	ret = ltdb_search_dn1_index(module, dn_key, msg);
+	ret = ltdb_search_dn1_index(module, dn_key, msg, NULL);
+
 	if (ret != LDB_SUCCESS && ret != LDB_ERR_NO_SUCH_OBJECT) {
 		talloc_free(msg);
 		return ret;
@@ -1284,21 +1253,30 @@ static int ltdb_index_add1(struct ldb_module *module, const char *dn,
 		msg->num_elements = 0;
 		msg->elements = NULL;
 	}
+	
+	el = ldb_msg_find_element(msg, LTDB_IDX);
 
-	for (i=0;i<msg->num_elements;i++) {
-		if (strcmp(LTDB_IDX, msg->elements[i].name) == 0) {
-			break;
-		}
+	casefold_dn = ldb_dn_get_casefold_as_ldb_val(dn);
+	if (!casefold_dn.data) {
+		talloc_free(msg);
+		return LDB_ERR_INVALID_DN_SYNTAX;
 	}
 
-	if (i == msg->num_elements) {
-		ret = ltdb_index_add1_new(ldb, msg, dn);
+	casefold_dn.data = talloc_memdup(msg, casefold_dn.data, casefold_dn.length);
+	if (!casefold_dn.data) {
+		talloc_free(msg);
+		ldb_oom(ldb);
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	if (!el) {
+		ret = ltdb_index_add1_new(ldb, msg, &casefold_dn, &el);
 	} else {
-		ret = ltdb_index_add1_add(ldb, msg, i, dn, a);
+		ret = ltdb_index_add1_add(ldb, msg, el, &casefold_dn, a);
 	}
 
 	if (ret == LDB_SUCCESS) {
-		ret = ltdb_store_idxptr(module, msg, TDB_REPLACE);
+		ret = ltdb_store_idxptr(module, msg, msg, el, TDB_REPLACE);
 	}
 
 	talloc_free(msg);
@@ -1306,7 +1284,8 @@ static int ltdb_index_add1(struct ldb_module *module, const char *dn,
 	return ret;
 }
 
-static int ltdb_index_add0(struct ldb_module *module, const char *dn,
+static int ltdb_index_add0(struct ldb_module *module, TALLOC_CTX *mem_ctx, 
+			   struct ldb_dn *dn,
 			   struct ldb_message_element *elements, int num_el)
 {
 	void *data = ldb_module_get_private(module);
@@ -1314,10 +1293,6 @@ static int ltdb_index_add0(struct ldb_module *module, const char *dn,
 	int ret;
 	unsigned int i, j;
 
-	if (dn[0] == '@') {
-		return LDB_SUCCESS;
-	}
-
 	if (ltdb->cache->indexlist->num_elements == 0) {
 		/* no indexed fields */
 		return LDB_SUCCESS;
@@ -1325,12 +1300,12 @@ static int ltdb_index_add0(struct ldb_module *module, const char *dn,
 
 	for (i = 0; i < num_el; i++) {
 		ret = ldb_msg_find_idx(ltdb->cache->indexlist, elements[i].name,
-				       NULL, LTDB_IDXATTR);
+				       LTDB_IDXATTR);
 		if (ret == -1) {
 			continue;
 		}
 		for (j = 0; j < elements[i].num_values; j++) {
-			ret = ltdb_index_add1(module, dn, &elements[i], j);
+			ret = ltdb_index_add1(module, mem_ctx, dn, &elements[i], j);
 			if (ret != LDB_SUCCESS) {
 				return ret;
 			}
@@ -1343,17 +1318,17 @@ static int ltdb_index_add0(struct ldb_module *module, const char *dn,
 /*
   add the index entries for a new record
 */
-int ltdb_index_add(struct ldb_module *module, const struct ldb_message *msg)
+int ltdb_index_add(struct ldb_module *module, TALLOC_CTX *mem_ctx, 
+		   const struct ldb_message *msg)
 {
-	const char *dn;
 	int ret;
 
-	dn = ldb_dn_get_linearized(msg->dn);
-	if (dn == NULL) {
-		return LDB_ERR_OPERATIONS_ERROR;
+	if (ldb_dn_is_special(msg->dn)) {
+		return LDB_SUCCESS;
 	}
 
-	ret = ltdb_index_add0(module, dn, msg->elements, msg->num_elements);
+	ret = ltdb_index_add0(module, mem_ctx, 
+			      msg->dn, msg->elements, msg->num_elements);
 
 	return ret;
 }
@@ -1362,76 +1337,93 @@ int ltdb_index_add(struct ldb_module *module, const struct ldb_message *msg)
 /*
   delete an index entry for one message element
 */
-int ltdb_index_del_value(struct ldb_module *module, const char *dn,
+int ltdb_index_del_value(struct ldb_module *module, TALLOC_CTX *mem_ctx, 
+			 struct ldb_dn *dn,
 			 struct ldb_message_element *el, int v_idx)
 {
 	struct ldb_context *ldb;
 	struct ldb_message *msg;
 	struct ldb_dn *dn_key;
-	int ret, i;
-	unsigned int j;
+	struct ldb_val dn_as_ldb_val, *found_val;
+	int ret;
 
 	ldb = ldb_module_get_ctx(module);
 
-	if (dn[0] == '@') {
+	if (ldb_dn_is_special(dn)) {
 		return LDB_SUCCESS;
 	}
 
-	dn_key = ltdb_index_key(ldb, el->name, &el->values[v_idx], NULL);
-	if (!dn_key) {
+	msg = talloc(mem_ctx, struct ldb_message);
+	if (msg == NULL) {
+		ldb_oom(ldb);
 		return LDB_ERR_OPERATIONS_ERROR;
 	}
 
-	msg = talloc(dn_key, struct ldb_message);
-	if (msg == NULL) {
-		talloc_free(dn_key);
+	dn_key = ltdb_index_key(ldb, msg, el->name, &el->values[v_idx], NULL);
+	if (!dn_key) {
+		talloc_free(msg);
 		return LDB_ERR_OPERATIONS_ERROR;
 	}
 
-	ret = ltdb_search_dn1_index(module, dn_key, msg);
+	ret = ltdb_search_dn1_index(module, dn_key, msg, NULL);
+
 	if (ret != LDB_SUCCESS && ret != LDB_ERR_NO_SUCH_OBJECT) {
-		talloc_free(dn_key);
-		return ret;
+		talloc_free(msg);
+		return LDB_ERR_OPERATIONS_ERROR;
 	}
 
 	if (ret == LDB_ERR_NO_SUCH_OBJECT) {
+		talloc_free(msg);
 		/* it wasn't indexed. Did we have an earlier error? If we did then
 		   its gone now */
-		talloc_free(dn_key);
 		return LDB_SUCCESS;
 	}
 
-	i = ldb_msg_find_idx(msg, dn, &j, LTDB_IDX);
-	if (i == -1) {
+	el = ldb_msg_find_element(msg, LTDB_IDX);
+	if (!el) {
+		talloc_free(msg);
+		/* there was set of index values on this index. Did we have an earlier error? If we did then
+		   its gone now */
+		return LDB_SUCCESS;
+	}
+
+	
+	dn_as_ldb_val = ldb_dn_get_casefold_as_ldb_val(dn);
+	if (!dn_as_ldb_val.data) {
+		return LDB_ERR_INVALID_DN_SYNTAX;
+	}
+
+	found_val = ldb_msg_find_val(el, &dn_as_ldb_val);
+
+	if (!found_val) {
 		struct ldb_ldif ldif;
 		char *ldif_string;
 		ldif.changetype = LDB_CHANGETYPE_NONE;
 		ldif.msg = msg;
-		ldif_string = ldb_ldif_write_string(ldb, NULL, &ldif);
+		ldif_string = ldb_ldif_write_string(ldb, msg, &ldif);
 		ldb_debug(ldb, LDB_DEBUG_ERROR,
-			  "ERROR: dn %s not found in %s", dn,
+			  "ERROR: dn %s not found in %s", ldb_dn_get_linearized(dn),
 			  ldif_string);
-		talloc_free(ldif_string);
 		/* it ain't there. hmmm */
-		talloc_free(dn_key);
+		talloc_free(msg);
 		return LDB_SUCCESS;
 	}
 
-	if (j != msg->elements[i].num_values - 1) {
-		memmove(&msg->elements[i].values[j],
-			&msg->elements[i].values[j+1],
-			(msg->elements[i].num_values-(j+1)) *
-			sizeof(msg->elements[i].values[0]));
-	}
-	msg->elements[i].num_values--;
+	talloc_free(found_val->data);
 
-	if (msg->elements[i].num_values == 0) {
-		ret = ltdb_delete_noindex(module, dn_key);
+	if (el->num_values == 1) {
+		ret = ltdb_delete_noindex(module, msg, dn_key);
 	} else {
-		ret = ltdb_store_idxptr(module, msg, TDB_REPLACE);
+		int n = (found_val - el->values);
+		if (n != el->num_values-1) {
+			memmove(found_val, found_val+1, ((el->num_values-1) - n)*sizeof(*found_val));
+		}
+		el->num_values--;
+
+		ret = ltdb_store_idxptr(module, msg, msg, el, TDB_REPLACE);
 	}
 
-	talloc_free(dn_key);
+	talloc_free(msg);
 
 	return ret;
 }
@@ -1440,12 +1432,11 @@ int ltdb_index_del_value(struct ldb_module *module, const char *dn,
   delete the index entries for a record
   return -1 on failure
 */
-int ltdb_index_del(struct ldb_module *module, const struct ldb_message *msg)
+int ltdb_index_del(struct ldb_module *module, TALLOC_CTX *mem_ctx, const struct ldb_message *msg)
 {
 	void *data = ldb_module_get_private(module);
 	struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
 	int ret;
-	const char *dn;
 	unsigned int i, j;
 
 	/* find the list of indexed fields */
@@ -1458,19 +1449,14 @@ int ltdb_index_del(struct ldb_module *module, const struct ldb_message *msg)
 		return LDB_SUCCESS;
 	}
 
-	dn = ldb_dn_get_linearized(msg->dn);
-	if (dn == NULL) {
-		return LDB_ERR_OPERATIONS_ERROR;
-	}
-
 	for (i = 0; i < msg->num_elements; i++) {
 		ret = ldb_msg_find_idx(ltdb->cache->indexlist, msg->elements[i].name, 
-				       NULL, LTDB_IDXATTR);
+				       LTDB_IDXATTR);
 		if (ret == -1) {
 			continue;
 		}
 		for (j = 0; j < msg->elements[i].num_values; j++) {
-			ret = ltdb_index_del_value(module, dn, &msg->elements[i], j);
+			ret = ltdb_index_del_value(module, mem_ctx, msg->dn, &msg->elements[i], j);
 			if (ret != LDB_SUCCESS) {
 				return ret;
 			}
@@ -1483,14 +1469,14 @@ int ltdb_index_del(struct ldb_module *module, const struct ldb_message *msg)
 /*
   handle special index for one level searches
 */
-int ltdb_index_one(struct ldb_module *module, const struct ldb_message *msg, int add)
+int ltdb_index_one(struct ldb_module *module, TALLOC_CTX *mem_ctx, 
+		   const struct ldb_message *msg, int add)
 {
 	void *data = ldb_module_get_private(module);
 	struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
 	struct ldb_message_element el;
 	struct ldb_val val;
 	struct ldb_dn *pdn;
-	const char *dn;
 	int ret;
 
 	if (ldb_dn_is_special(msg->dn)) {
@@ -1498,37 +1484,30 @@ int ltdb_index_one(struct ldb_module *module, const struct ldb_message *msg, int
 	}
 
 	/* We index for ONE Level only if requested */
-	ret = ldb_msg_find_idx(ltdb->cache->indexlist, NULL, NULL, LTDB_IDXONE);
+	ret = ldb_msg_find_idx(ltdb->cache->indexlist, NULL, LTDB_IDXONE);
 	if (ret != 0) {
 		return LDB_SUCCESS;
 	}
 
-	pdn = ldb_dn_get_parent(module, msg->dn);
+	pdn = ldb_dn_get_parent(mem_ctx, msg->dn);
 	if (pdn == NULL) {
 		return LDB_ERR_OPERATIONS_ERROR;
 	}
 
-	dn = ldb_dn_get_linearized(msg->dn);
-	if (dn == NULL) {
-		talloc_free(pdn);
-		return LDB_ERR_OPERATIONS_ERROR;
-	}
-
-	val.data = (uint8_t *)((uintptr_t)ldb_dn_get_casefold(pdn));
+	val = ldb_dn_get_casefold_as_ldb_val(pdn);
 	if (val.data == NULL) {
 		talloc_free(pdn);
-		return LDB_ERR_OPERATIONS_ERROR;
+		return LDB_ERR_INVALID_DN_SYNTAX;
 	}
 
-	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);
+		ret = ltdb_index_add1(module, pdn, msg->dn, &el, 0);
 	} else { /* delete */
-		ret = ltdb_index_del_value(module, dn, &el, 0);
+		ret = ltdb_index_del_value(module, pdn, msg->dn, &el, 0);
 	}
 
 	talloc_free(pdn);
@@ -1557,7 +1536,6 @@ static int re_index(struct tdb_context *tdb, TDB_DATA key, TDB_DATA data, void *
 	struct ldb_context *ldb;
 	struct ldb_module *module = (struct ldb_module *)state;
 	struct ldb_message *msg;
-	const char *dn = NULL;
 	int ret;
 	TDB_DATA key2;
 
@@ -1583,7 +1561,7 @@ static int re_index(struct tdb_context *tdb, TDB_DATA key, TDB_DATA data, void *
 
 	/* check if the DN key has changed, perhaps due to the
 	   case insensitivity of an element changing */
-	key2 = ltdb_key(module, msg->dn);
+	key2 = ltdb_key(msg, msg->dn);
 	if (key2.dptr == NULL) {
 		/* probably a corrupt record ... darn */
 		ldb_debug(ldb, LDB_DEBUG_ERROR, "Invalid DN in re_index: %s",
@@ -1591,21 +1569,15 @@ static int re_index(struct tdb_context *tdb, TDB_DATA key, TDB_DATA data, void *
 		talloc_free(msg);
 		return 0;
 	}
-	if (strcmp((char *)key2.dptr, (char *)key.dptr) != 0) {
+
+	if (tdb_data_cmp(&key2, &key) != 0) {
 		tdb_delete(tdb, key);
 		tdb_store(tdb, key2, data, 0);
 	}
-	talloc_free(key2.dptr);
 
-	if (msg->dn == NULL) {
-		dn = (char *)key.dptr + 3;
-	} else {
-		dn = ldb_dn_get_linearized(msg->dn);
-	}
-
-	ret = ltdb_index_one(module, msg, 1);
+	ret = ltdb_index_one(module, msg, msg, 1);
 	if (ret == LDB_SUCCESS) {
-		ret = ltdb_index_add0(module, dn, msg->elements, msg->num_elements);
+		ret = ltdb_index_add0(module, msg, msg->dn, msg->elements, msg->num_elements);
 	} else {
 		ldb_debug(ldb, LDB_DEBUG_ERROR,
 			"Adding special ONE LEVEL index failed (%s)!",
@@ -1653,5 +1625,5 @@ int ltdb_reindex(struct ldb_module *module)
 		ltdb->idxptr->repack = true;
 	}
 
-	return LDB_SUCCESS;
+	return ltdb_set_casefold_index(module);
 }
diff --git a/source4/lib/ldb/ldb_tdb/ldb_search.c b/source4/lib/ldb/ldb_tdb/ldb_search.c
index a089a2f826..a128d9cc0c 100644
--- a/source4/lib/ldb/ldb_tdb/ldb_search.c
+++ b/source4/lib/ldb/ldb_tdb/ldb_search.c
@@ -232,29 +232,23 @@ static int ltdb_search_base(struct ldb_module *module, struct ldb_dn *dn)
 }
 
 /*
-  search the database for a single simple dn, returning all attributes
+  search the database for a single tdb key, returning all attributes
   in a single message
 
   return LDB_ERR_NO_SUCH_OBJECT on record-not-found
   and LDB_SUCCESS on success
 */
-int ltdb_search_dn1(struct ldb_module *module, struct ldb_dn *dn, struct ldb_message *msg)
+int ltdb_search_dn1_key(struct ldb_module *module, 
+			TDB_DATA tdb_key, struct ldb_message *msg)
 {
 	void *data = ldb_module_get_private(module);
 	struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
 	int ret;
-	TDB_DATA tdb_key, tdb_data;
+	TDB_DATA tdb_data;
 
 	memset(msg, 0, sizeof(*msg));
 
-	/* form the key */
-	tdb_key = ltdb_key(module, dn);
-	if (!tdb_key.dptr) {
-		return LDB_ERR_OPERATIONS_ERROR;
-	}
-
 	tdb_data = tdb_fetch(ltdb->tdb, tdb_key);
-	talloc_free(tdb_key.dptr);
 	if (!tdb_data.dptr) {
 		return LDB_ERR_NO_SUCH_OBJECT;
 	}
@@ -272,9 +266,6 @@ int ltdb_search_dn1(struct ldb_module *module, struct ldb_dn *dn, struct ldb_mes
 	}
 
 	if (!msg->dn) {
-		msg->dn = ldb_dn_copy(msg, dn);
-	}
-	if (!msg->dn) {
 		return LDB_ERR_OPERATIONS_ERROR;
 	}
 
@@ -282,6 +273,32 @@ int ltdb_search_dn1(struct ldb_module *module, struct ldb_dn *dn, struct ldb_mes
 }
 
 /*
+  search the database for a single simple dn, returning all attributes
+  in a single message
+
+  return LDB_ERR_NO_SUCH_OBJECT on record-not-found
+  and LDB_SUCCESS on success
+*/
+
+int ltdb_search_dn1(struct ldb_module *module, struct ldb_dn *dn, struct ldb_message *msg)
+{
+	int ret;
+	TDB_DATA tdb_key;
+
+	memset(msg, 0, sizeof(*msg));
+
+	/* form the key */
+	tdb_key = ltdb_key(msg, dn);
+	if (!tdb_key.dptr) {
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	ret = ltdb_search_dn1_key(module, tdb_key, msg);
+	talloc_free(tdb_key.dptr);
+	return ret;
+}
+
+/*
   add a set of attributes from a record to a set of results
   return 0 on success, -1 on failure
 */
diff --git a/source4/lib/ldb/ldb_tdb/ldb_tdb.c b/source4/lib/ldb/ldb_tdb/ldb_tdb.c
index 55acb6132d..2348c0dcce 100644
--- a/source4/lib/ldb/ldb_tdb/ldb_tdb.c
+++ b/source4/lib/ldb/ldb_tdb/ldb_tdb.c
@@ -103,6 +103,53 @@ int ltdb_unlock_read(struct ldb_module *module)
 	return 0;
 }
 
+struct ldb_val ldb_dn_get_casefold_as_ldb_val(struct ldb_dn *dn) {
+	struct ldb_val val;
+	const char *casefold_dn = ldb_dn_get_casefold(dn);
+	val.data = (uint8_t *)((uintptr_t)casefold_dn);
+	val.length = strlen(casefold_dn);
+	return val;
+}
+
+struct ldb_val ldb_dn_alloc_casefold_as_ldb_val(TALLOC_CTX *mem_ctx, struct ldb_dn *dn) {
+	struct ldb_val val;
+	const char *casefold_dn = ldb_dn_alloc_casefold(mem_ctx, dn);
+	val.data = (uint8_t *)((uintptr_t)casefold_dn);
+	val.length = strlen(casefold_dn);
+	return val;
+}
+
+/*
+  form a TDB_DATA for a record key
+  caller frees
+
+  This version takes the casefolded string form of the DN as an ldb_val
+*/
+struct TDB_DATA ltdb_key_from_casefold_dn(TALLOC_CTX *mem_ctx, 
+					  struct ldb_val dn_folded)
+{
+	TDB_DATA key;
+
+	key.dsize = dn_folded.length + 4;
+	key.dptr = talloc_size(mem_ctx, key.dsize);
+	if (!key.dptr) {
+		goto failed;
+	}
+
+	memcpy(key.dptr, "DN=", 3);
+	memcpy(&key.dptr[3], dn_folded.data, key.dsize - 4);
+
+	key.dptr[key.dsize - 1] = '\0';
+
+	return key;
+
+failed:
+	errno = ENOMEM;
+	key.dptr = NULL;
+	key.dsize = 0;
+	return key;
+}
+
 
 /*
   form a TDB_DATA for a record key
@@ -111,12 +158,10 @@ int ltdb_unlock_read(struct ldb_module *module)
   note that the key for a record can depend on whether the
   dn refers to a case sensitive index record or not
 */
-struct TDB_DATA ltdb_key(struct ldb_module *module, struct ldb_dn *dn)
+struct TDB_DATA ltdb_key(TALLOC_CTX *mem_ctx, struct ldb_dn *dn)
 {
-	struct ldb_context *ldb = ldb_module_get_ctx(module);
 	TDB_DATA key;
-	char *key_str = NULL;
-	const char *dn_folded = NULL;
+	struct ldb_val dn_folded;
 
 	/*
 	  most DNs are case insensitive. The exception is index DNs for
@@ -130,31 +175,15 @@ struct TDB_DATA ltdb_key(struct ldb_module *module, struct ldb_dn *dn)
 	     the indexing code handles the rest
 	*/
 
-	dn_folded = ldb_dn_get_casefold(dn);
-	if (!dn_folded) {
-		goto failed;
-	}
-
-	key_str = talloc_strdup(ldb, "DN=");
-	if (!key_str) {
-		goto failed;
-	}
-
-	key_str = talloc_strdup_append_buffer(key_str, dn_folded);
-	if (!key_str) {
-		goto failed;
+	dn_folded = ldb_dn_get_casefold_as_ldb_val(dn);
+	if (!dn_folded.data) {
+		errno = EINVAL;
+		key.dptr = NULL;
+		key.dsize = 0;
+		return key;
 	}
 
-	key.dptr = (uint8_t *)key_str;
-	key.dsize = strlen(key_str) + 1;
-
-	return key;
-
-failed:
-	errno = ENOMEM;
-	key.dptr = NULL;
-	key.dsize = 0;
-	return key;
+	return ltdb_key_from_casefold_dn(mem_ctx, dn_folded);
 }
 
 /*
@@ -213,14 +242,15 @@ static int ltdb_modified(struct ldb_module *module, struct ldb_dn *dn)
 /*
   store a record into the db
 */
-int ltdb_store(struct ldb_module *module, const struct ldb_message *msg, int flgs)
+int ltdb_store(struct ldb_module *module, TALLOC_CTX *mem_ctx, 
+	       const struct ldb_message *msg, int flgs)
 {
 	void *data = ldb_module_get_private(module);
 	struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
 	TDB_DATA tdb_key, tdb_data;
 	int ret;
 
-	tdb_key = ltdb_key(module, msg->dn);
+	tdb_key = ltdb_key(mem_ctx, msg->dn);
 	if (!tdb_key.dptr) {
 		return LDB_ERR_OTHER;
 	}
@@ -237,7 +267,7 @@ int ltdb_store(struct ldb_module *module, const struct ldb_message *msg, int flg
 		goto done;
 	}
 
-	ret = ltdb_index_add(module, msg);
+	ret = ltdb_index_add(module, mem_ctx, msg);
 	if (ret != LDB_SUCCESS) {
 		tdb_delete(ltdb->tdb, tdb_key);
 	}
@@ -251,6 +281,7 @@ done:
 
 
 static int ltdb_add_internal(struct ldb_module *module,
+			     TALLOC_CTX *mem_ctx, 
 			     const struct ldb_message *msg)
 {
 	struct ldb_context *ldb = ldb_module_get_ctx(module);
@@ -283,7 +314,7 @@ static int ltdb_add_internal(struct ldb_module *module,
 		}
 	}
 
-	ret = ltdb_store(module, msg, TDB_INSERT);
+	ret = ltdb_store(module, mem_ctx, msg, TDB_INSERT);
 
 	if (ret == LDB_ERR_ENTRY_ALREADY_EXISTS) {
 		ldb_asprintf_errstring(ldb,
@@ -293,7 +324,7 @@ static int ltdb_add_internal(struct ldb_module *module,
 	}
 
 	if (ret == LDB_SUCCESS) {
-		ret = ltdb_index_one(module, msg, 1);
+		ret = ltdb_index_one(module, mem_ctx, msg, 1);
 		if (ret != LDB_SUCCESS) {
 			return ret;
 		}
@@ -318,7 +349,7 @@ static int ltdb_add(struct ltdb_context *ctx)
 
 	ldb_request_set_state(req, LDB_ASYNC_PENDING);
 
-	tret = ltdb_add_internal(module, req->op.add.message);
+	tret = ltdb_add_internal(module, req, req->op.add.message);
 	if (tret != LDB_SUCCESS) {
 		return tret;
 	}
@@ -330,14 +361,14 @@ static int ltdb_add(struct ltdb_context *ctx)
   delete a record from the database, not updating indexes (used for deleting
   index records)
 */
-int ltdb_delete_noindex(struct ldb_module *module, struct ldb_dn *dn)
+int ltdb_delete_noindex(struct ldb_module *module, TALLOC_CTX *mem_ctx, struct ldb_dn *dn)
 {
 	void *data = ldb_module_get_private(module);
 	struct ltdb_private *ltdb = talloc_get_type(data, struct ltdb_private);
 	TDB_DATA tdb_key;
 	int ret;
 
-	tdb_key = ltdb_key(module, dn);
+	tdb_key = ltdb_key(mem_ctx, dn);
 	if (!tdb_key.dptr) {
 		return LDB_ERR_OTHER;
 	}
@@ -352,12 +383,12 @@ int ltdb_delete_noindex(struct ldb_module *module, struct ldb_dn *dn)
 	return ret;
 }
 
-static int ltdb_delete_internal(struct ldb_module *module, struct ldb_dn *dn)
+static int ltdb_delete_internal(struct ldb_module *module, TALLOC_CTX *mem_ctx, struct ldb_dn *dn)
 {
 	struct ldb_message *msg;
 	int ret;
 
-	msg = talloc(module, struct ldb_message);
+	msg = talloc(mem_ctx, struct ldb_message);
 	if (msg == NULL) {
 		return LDB_ERR_OPERATIONS_ERROR;
 	}
@@ -370,19 +401,19 @@ static int ltdb_delete_internal(struct ldb_module *module, struct ldb_dn *dn)
 		goto done;
 	}
 
-	ret = ltdb_delete_noindex(module, dn);
+	ret = ltdb_delete_noindex(module, msg, dn);
 	if (ret != LDB_SUCCESS) {
 		goto done;
 	}
 
 	/* remove one level attribute */
-	ret = ltdb_index_one(module, msg, 0);
+	ret = ltdb_index_one(module, msg, msg, 0);
 	if (ret != LDB_SUCCESS) {
 		goto done;
 	}
 
 	/* remove any indexed attributes */
-	ret = ltdb_index_del(module, msg);
+	ret = ltdb_index_del(module, msg, msg);
 	if (ret != LDB_SUCCESS) {
 		goto done;
 	}
@@ -412,7 +443,7 @@ static int ltdb_delete(struct ltdb_context *ctx)
 		return LDB_ERR_OPERATIONS_ERROR;
 	}
 
-	tret = ltdb_delete_internal(module, req->op.del.dn);
+	tret = ltdb_delete_internal(module, req, req->op.del.dn);
 	if (tret != LDB_SUCCESS) {
 		return tret;
 	}
@@ -489,21 +520,14 @@ static int msg_add_element(struct ldb_context *ldb,
   delete all elements having a specified attribute name
 */
 static int msg_delete_attribute(struct ldb_module *module,
-				struct ldb_context *ldb,
 				struct ldb_message *msg, const char *name)
 {
-	const char *dn;
 	unsigned int i, j;
 
-	dn = ldb_dn_get_linearized(msg->dn);
-	if (dn == NULL) {
-		return -1;
-	}
-
 	for (i=0;i<msg->num_elements;i++) {
 		if (ldb_attr_cmp(msg->elements[i].name, name) == 0) {
 			for (j=0;j<msg->elements[i].num_values;j++) {
-				ltdb_index_del_value(module, dn,
+				ltdb_index_del_value(module, msg, msg->dn,
 						     &msg->elements[i], j);
 			}
 			talloc_free(msg->elements[i].values);
@@ -550,7 +574,7 @@ static int msg_delete_element(struct ldb_module *module,
 	a = ldb_schema_attribute_by_name(ldb, el->name);
 
 	for (i=0;i<el->num_values;i++) {
-		if (a->syntax->comparison_fn(ldb, ldb,
+		if (a->syntax->comparison_fn(ldb, msg,
 						&el->values[i], val) == 0) {
 			if (i<el->num_values-1) {
 				memmove(&el->values[i], &el->values[i+1],
@@ -559,7 +583,7 @@ static int msg_delete_element(struct ldb_module *module,
 			}
 			el->num_values--;
 			if (el->num_values == 0) {
-				return msg_delete_attribute(module, ldb,
+				return msg_delete_attribute(module, 
 							    msg, name);
 			}
 			return 0;
@@ -578,6 +602,7 @@ static int msg_delete_element(struct ldb_module *module,
   then we'll need to look at this again
 */
 int ltdb_modify_internal(struct ldb_module *module,
+			 TALLOC_CTX *mem_ctx, 
 			 const struct ldb_message *msg)
 {
 	struct ldb_context *ldb = ldb_module_get_ctx(module);
@@ -587,28 +612,32 @@ int ltdb_modify_internal(struct ldb_module *module,
 	struct ldb_message *msg2;
 	unsigned i, j;
 	int ret, idx;
-
-	tdb_key = ltdb_key(module, msg->dn);
+	TALLOC_CTX *tmp_ctx = talloc_new(mem_ctx);
+	if (!tmp_ctx) {
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+	tdb_key = ltdb_key(tmp_ctx, msg->dn);
 	if (!tdb_key.dptr) {
+		talloc_free(tmp_ctx);
 		return LDB_ERR_OTHER;
 	}
 
 	tdb_data = tdb_fetch(ltdb->tdb, tdb_key);
+	talloc_free(tdb_key.dptr);
+
 	if (!tdb_data.dptr) {
-		talloc_free(tdb_key.dptr);
 		return ltdb_err_map(tdb_error(ltdb->tdb));
 	}
 
-	msg2 = talloc(tdb_key.dptr, struct ldb_message);
+	msg2 = talloc(tmp_ctx, struct ldb_message);
 	if (msg2 == NULL) {
-		talloc_free(tdb_key.dptr);
-		return LDB_ERR_OTHER;
+		ldb_oom(ldb);
+		ret = LDB_ERR_OPERATIONS_ERROR;
+		goto failed;
 	}
 
 	ret = ltdb_unpack_data(module, &tdb_data, msg2);
 	if (ret == -1) {
-		ret = LDB_ERR_OTHER;
-		goto failed;
 	}
 
 	if (!msg2->dn) {
@@ -619,7 +648,6 @@ int ltdb_modify_internal(struct ldb_module *module,
 		struct ldb_message_element *el = &msg->elements[i];
 		struct ldb_message_element *el2;
 		struct ldb_val *vals;
-		const char *dn;
 		const struct ldb_schema_attribute *a = ldb_schema_attribute_by_name(ldb, el->name);
 		switch (msg->elements[i].flags & LDB_FLAG_MOD_MASK) {
 
@@ -680,7 +708,8 @@ int ltdb_modify_internal(struct ldb_module *module,
 						el2->num_values + el->num_values);
 
 			if (vals == NULL) {
-				ret = LDB_ERR_OTHER;
+				ldb_oom(ldb);
+				ret = LDB_ERR_OPERATIONS_ERROR;
 				goto failed;
 			}
 
@@ -704,7 +733,7 @@ int ltdb_modify_internal(struct ldb_module *module,
 			}
 			/* replace all elements of this attribute name with the elements
 			   listed. The attribute not existing is not an error */
-			msg_delete_attribute(module, ldb, msg2, el->name);
+			msg_delete_attribute(module, msg2, el->name);
 
 			for (j=0;j<el->num_values;j++) {
 				if (ldb_msg_find_val(el, &el->values[j]) != &el->values[j]) {
@@ -724,17 +753,12 @@ int ltdb_modify_internal(struct ldb_module *module,
 
 		case LDB_FLAG_MOD_DELETE:
 
-			dn = ldb_dn_get_linearized(msg->dn);
-			if (dn == NULL) {
-				ret = LDB_ERR_OTHER;
-				goto failed;
-			}
-
 			/* we could be being asked to delete all
 			   values or just some values */
 			if (msg->elements[i].num_values == 0) {
-				if (msg_delete_attribute(module, ldb, msg2, 
+				if (msg_delete_attribute(module, msg2, 
 							 msg->elements[i].name) != 0) {
+					const char *dn = ldb_dn_get_linearized(msg->dn);
 					ldb_asprintf_errstring(ldb, "No such attribute: %s for delete on %s", msg->elements[i].name, dn);
 					ret = LDB_ERR_NO_SUCH_ATTRIBUTE;
 					goto failed;
@@ -746,11 +770,15 @@ int ltdb_modify_internal(struct ldb_module *module,
 						       msg2, 
 						       msg->elements[i].name,
 						       &msg->elements[i].values[j]) != 0) {
-					ldb_asprintf_errstring(ldb, "No matching attribute value when deleting attribute: %s on %s", msg->elements[i].name, dn);
+					const char *dn = ldb_dn_get_linearized(msg->dn);
+					ldb_asprintf_errstring(ldb, "No matching attribute value (%*.*s) when deleting attribute: %s on %s", 
+							       (int)msg->elements[i].values[j].length, (int)msg->elements[i].values[j].length, 
+							       (const char *)msg->elements[i].values[j].data,
+							       msg->elements[i].name, dn);
 					ret = LDB_ERR_NO_SUCH_ATTRIBUTE;
 					goto failed;
 				}
-				ret = ltdb_index_del_value(module, dn, &msg->elements[i], j);
+				ret = ltdb_index_del_value(module, tmp_ctx, msg->dn, &msg->elements[i], j);
 				if (ret != LDB_SUCCESS) {
 					goto failed;
 				}
@@ -768,7 +796,7 @@ int ltdb_modify_internal(struct ldb_module *module,
 
 	/* we've made all the mods
 	 * save the modified record back into the database */
-	ret = ltdb_store(module, msg2, TDB_MODIFY);
+	ret = ltdb_store(module, mem_ctx, msg2, TDB_MODIFY);
 	if (ret != LDB_SUCCESS) {
 		goto failed;
 	}
@@ -778,12 +806,11 @@ int ltdb_modify_internal(struct ldb_module *module,
 		goto failed;
 	}
 
-	talloc_free(tdb_key.dptr);
 	free(tdb_data.dptr);
 	return ret;
 
 failed:
-	talloc_free(tdb_key.dptr);
+	talloc_free(tmp_ctx);
 	free(tdb_data.dptr);
 	return ret;
 }
@@ -808,7 +835,7 @@ static int ltdb_modify(struct ltdb_context *ctx)
 		return LDB_ERR_OPERATIONS_ERROR;
 	}
 
-	tret = ltdb_modify_internal(module, req->op.mod.message);
+	tret = ltdb_modify_internal(module, req, req->op.mod.message);
 	if (tret != LDB_SUCCESS) {
 		return tret;
 	}
@@ -841,12 +868,14 @@ static int ltdb_rename(struct ltdb_context *ctx)
 	   to fetch the old record */
 	tret = ltdb_search_dn1(module, req->op.rename.olddn, msg);
 	if (tret != LDB_SUCCESS) {
+		talloc_free(msg);
 		/* not finding the old record is an error */
 		return tret;
 	}
 
 	msg->dn = ldb_dn_copy(msg, req->op.rename.newdn);
 	if (!msg->dn) {
+		talloc_free(msg);
 		return LDB_ERR_OPERATIONS_ERROR;
 	}
 
@@ -854,12 +883,14 @@ static int ltdb_rename(struct ltdb_context *ctx)
 	 * unique indexes. We rely on the transaction to make this
 	 * atomic
 	 */
-	tret = ltdb_delete_internal(module, req->op.rename.olddn);
+	tret = ltdb_delete_internal(module, msg, req->op.rename.olddn);
 	if (tret != LDB_SUCCESS) {
+		talloc_free(msg);
 		return tret;
 	}
 
-	tret = ltdb_add_internal(module, msg);
+	tret = ltdb_add_internal(module, msg, msg);
+	talloc_free(msg);
 	if (tret != LDB_SUCCESS) {
 		return tret;
 	}
@@ -892,7 +923,7 @@ static int ltdb_prepare_commit(struct ldb_module *module)
 		return LDB_SUCCESS;
 	}
 
-	if (ltdb_index_transaction_commit(module) != 0) {
+	if (ltdb_index_transaction_prepare_commit(module) != 0) {
 		tdb_transaction_cancel(ltdb->tdb);
 		ltdb->in_transaction--;
 		return ltdb_err_map(tdb_error(ltdb->tdb));
diff --git a/source4/lib/ldb/ldb_tdb/ldb_tdb.h b/source4/lib/ldb/ldb_tdb/ldb_tdb.h
index c8c1dad5de..43f2909008 100644
--- a/source4/lib/ldb/ldb_tdb/ldb_tdb.h
+++ b/source4/lib/ldb/ldb_tdb/ldb_tdb.h
@@ -29,6 +29,8 @@ struct ltdb_private {
 	bool check_base;
 	struct ltdb_idxptr *idxptr;
 	bool prepared_commit;
+
+	int index_version;
 };
 
 /*
@@ -65,6 +67,14 @@ struct ltdb_context {
 #define LTDB_OPTIONS    "@OPTIONS"
 #define LTDB_ATTRIBUTES "@ATTRIBUTES"
 
+#define LTDB_INDEX_VERSION   "@INDEX_VERSION"
+
+/* ltdb index versions: 
+   0 - Initial version, DN values as index values, not casefolded
+   1 - DN values as index values, casefolded and sorted (binary compare)
+ */
+
+
 /* special attribute types */
 #define LTDB_SEQUENCE_NUMBER "sequenceNumber"
 #define LTDB_CHECK_BASE "checkBaseOnSearch"
@@ -76,6 +86,7 @@ struct ltdb_context {
 int ltdb_cache_reload(struct ldb_module *module);
 int ltdb_cache_load(struct ldb_module *module);
 int ltdb_increase_sequence_number(struct ldb_module *module);
+int ltdb_set_casefold_index(struct ldb_module *module);
 int ltdb_check_at_attributes_values(const struct ldb_val *value);
 
 /* The following definitions come from lib/ldb/ldb_tdb/ldb_index.c  */
@@ -83,12 +94,15 @@ int ltdb_check_at_attributes_values(const struct ldb_val *value);
 struct ldb_parse_tree;
 
 int ltdb_search_indexed(struct ltdb_context *ctx, uint32_t *);
-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_index_add(struct ldb_module *module, TALLOC_CTX *mem_ctx, 
+		   const struct ldb_message *msg);
+int ltdb_index_del(struct ldb_module *module, TALLOC_CTX *mem_ctx, 
+		   const struct ldb_message *msg);
+int ltdb_index_one(struct ldb_module *module, TALLOC_CTX *mem_ctx, 
+		   const struct ldb_message *msg, int add);
 int ltdb_reindex(struct ldb_module *module);
 int ltdb_index_transaction_start(struct ldb_module *module);
-int ltdb_index_transaction_commit(struct ldb_module *module);
+int ltdb_index_transaction_prepare_commit(struct ldb_module *module);
 int ltdb_index_transaction_cancel(struct ldb_module *module);
 
 /* The following definitions come from lib/ldb/ldb_tdb/ldb_pack.c  */
@@ -107,6 +121,14 @@ int ltdb_unpack_data(struct ldb_module *module,
 int ltdb_has_wildcard(struct ldb_module *module, const char *attr_name, 
 		      const struct ldb_val *val);
 void ltdb_search_dn1_free(struct ldb_module *module, struct ldb_message *msg);
+/*
+  search the database for a single tdb key, returning all attributes
+  in a single message
+
+  return LDB_ERR_NO_SUCH_OBJECT on record-not-found
+  and LDB_SUCCESS on success
+*/
+int ltdb_search_dn1_key(struct ldb_module *module, TDB_DATA tdb_key, struct ldb_message *msg);
 int ltdb_search_dn1(struct ldb_module *module, struct ldb_dn *dn, struct ldb_message *msg);
 int ltdb_add_attr_results(struct ldb_module *module,
  			  TALLOC_CTX *mem_ctx, 
@@ -120,12 +142,26 @@ int ltdb_search(struct ltdb_context *ctx);
 /* The following definitions come from lib/ldb/ldb_tdb/ldb_tdb.c  */
 int ltdb_lock_read(struct ldb_module *module);
 int ltdb_unlock_read(struct ldb_module *module);
-struct TDB_DATA ltdb_key(struct ldb_module *module, struct ldb_dn *dn);
-int ltdb_store(struct ldb_module *module, const struct ldb_message *msg, int flgs);
-int ltdb_delete_noindex(struct ldb_module *module, struct ldb_dn *dn);
-int ltdb_modify_internal(struct ldb_module *module, const struct ldb_message *msg);
+struct TDB_DATA ltdb_key(TALLOC_CTX *mem_ctx, struct ldb_dn *dn);
+/*
+  form a TDB_DATA for a record key
+  caller frees
 
-int ltdb_index_del_value(struct ldb_module *module, const char *dn, 
+  This version takes the casefolded string form of the DN as an ldb_val
+*/
+struct TDB_DATA ltdb_key_from_casefold_dn(TALLOC_CTX *mem_ctx, 
+					  struct ldb_val dn_folded);
+struct ldb_val ldb_dn_get_casefold_as_ldb_val(struct ldb_dn *dn);
+struct ldb_val ldb_dn_alloc_casefold_as_ldb_val(TALLOC_CTX *mem_ctx, struct ldb_dn *dn);
+int ltdb_store(struct ldb_module *module, TALLOC_CTX *mem_ctx, 
+	       const struct ldb_message *msg, int flgs);
+int ltdb_delete_noindex(struct ldb_module *module, TALLOC_CTX *mem_ctx, 
+			struct ldb_dn *dn);
+int ltdb_modify_internal(struct ldb_module *module, TALLOC_CTX *mem_ctx, 
+			 const struct ldb_message *msg);
+
+int ltdb_index_del_value(struct ldb_module *module, TALLOC_CTX *mem_ctx, 
+			 struct ldb_dn *dn,
 			 struct ldb_message_element *el, int v_idx);
 
 struct tdb_context *ltdb_wrap_open(TALLOC_CTX *mem_ctx,