Adding upstream version 2:4.20.0+dfsg.upstream/2%4.20.0+dfsg

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

5 files changed, 8145 insertions, 0 deletions

diff --git a/lib/ldb/ldb_key_value/ldb_kv.c b/lib/ldb/ldb_key_value/ldb_kv.c
new file mode 100644
index 0000000..ef69e5e
--- /dev/null
+++ b/lib/ldb/ldb_key_value/ldb_kv.c
@@ -0,0 +1,2291 @@
+/*
+   ldb database library
+
+   Copyright (C) Andrew Tridgell 2004
+   Copyright (C) Stefan Metzmacher 2004
+   Copyright (C) Simo Sorce 2006-2008
+   Copyright (C) Matthias Dieter Wallnöfer 2009-2010
+
+     ** NOTE! The following LGPL license applies to the ldb
+     ** library. This does NOT imply that all of Samba is released
+     ** under the LGPL
+
+   This library is free software; you can redistribute it and/or
+   modify it under the terms of the GNU Lesser General Public
+   License as published by the Free Software Foundation; either
+   version 3 of the License, or (at your option) any later version.
+
+   This library is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+   Lesser General Public License for more details.
+
+   You should have received a copy of the GNU Lesser General Public
+   License along with this library; if not, see <http://www.gnu.org/licenses/>.
+*/
+
+/*
+ *  Name: ldb_kv
+ *
+ *  Component: ldb key value backend
+ *
+ *  Description: core functions for ldb key value backend
+ *
+ *  Author: Andrew Tridgell
+ *  Author: Stefan Metzmacher
+ *
+ *  Modifications:
+ *
+ *  - description: make the module use asynchronous calls
+ *    date: Feb 2006
+ *    Author: Simo Sorce
+ *
+ *  - description: make it possible to use event contexts
+ *    date: Jan 2008
+ *    Author: Simo Sorce
+ *
+ *  - description: fix up memory leaks and small bugs
+ *    date: Oct 2009
+ *    Author: Matthias Dieter Wallnöfer
+ */
+
+#include "ldb_kv.h"
+#include "ldb_private.h"
+#include "lib/util/attr.h"
+
+/*
+  prevent memory errors on callbacks
+*/
+struct ldb_kv_req_spy {
+	struct ldb_kv_context *ctx;
+};
+
+/*
+ * Determine if this key could hold a record.  We allow the new GUID
+ * index, the old DN index and a possible future ID=
+ */
+bool ldb_kv_key_is_normal_record(struct ldb_val key)
+{
+	if (key.length < 4) {
+		return false;
+	}
+
+	/*
+	 * @ records are not normal records, we don't want to index
+	 * them nor search on them
+	 */
+	if (key.length > 4 &&
+	    memcmp(key.data, "DN=@", 4) == 0) {
+		return false;
+	}
+
+	/* All other DN= records are however */
+	if (memcmp(key.data, "DN=", 3) == 0) {
+		return true;
+	}
+
+	if (memcmp(key.data, "ID=", 3) == 0) {
+		return true;
+	}
+
+	if (key.length < sizeof(LDB_KV_GUID_KEY_PREFIX)) {
+		return false;
+	}
+
+	if (memcmp(key.data, LDB_KV_GUID_KEY_PREFIX,
+		   sizeof(LDB_KV_GUID_KEY_PREFIX) - 1) == 0) {
+		return true;
+	}
+
+	return false;
+}
+
+/*
+  form a ldb_val for a record key
+  caller frees
+
+  note that the key for a record can depend on whether the
+  dn refers to a case sensitive index record or not
+*/
+struct ldb_val ldb_kv_key_dn(TALLOC_CTX *mem_ctx,
+			     struct ldb_dn *dn)
+{
+	struct ldb_val key;
+	char *key_str = NULL;
+	const char *dn_folded = NULL;
+
+	/*
+	  most DNs are case insensitive. The exception is index DNs for
+	  case sensitive attributes
+
+	  there are 3 cases dealt with in this code:
+
+	  1) if the dn doesn't start with @ then uppercase the attribute
+             names and the attributes values of case insensitive attributes
+	  2) if the dn starts with @ then leave it alone -
+	     the indexing code handles the rest
+	*/
+
+	dn_folded = ldb_dn_get_casefold(dn);
+	if (!dn_folded) {
+		goto failed;
+	}
+
+	key_str = talloc_strdup(mem_ctx, "DN=");
+	if (!key_str) {
+		goto failed;
+	}
+
+	key_str = talloc_strdup_append_buffer(key_str, dn_folded);
+	if (!key_str) {
+		goto failed;
+	}
+
+	key.data = (uint8_t *)key_str;
+	key.length = strlen(key_str) + 1;
+
+	return key;
+
+failed:
+	errno = ENOMEM;
+	key.data = NULL;
+	key.length = 0;
+	return key;
+}
+
+/* The caller is to provide a correctly sized key */
+int ldb_kv_guid_to_key(const struct ldb_val *GUID_val,
+		       struct ldb_val *key)
+{
+	const char *GUID_prefix = LDB_KV_GUID_KEY_PREFIX;
+	const int GUID_prefix_len = sizeof(LDB_KV_GUID_KEY_PREFIX) - 1;
+
+	if (key->length != (GUID_val->length+GUID_prefix_len)) {
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	memcpy(key->data, GUID_prefix, GUID_prefix_len);
+	memcpy(&key->data[GUID_prefix_len],
+	       GUID_val->data, GUID_val->length);
+	return LDB_SUCCESS;
+}
+
+/*
+ * The caller is to provide a correctly sized key, used only in
+ * the GUID index mode
+ */
+int ldb_kv_idx_to_key(struct ldb_module *module,
+		      struct ldb_kv_private *ldb_kv,
+		      TALLOC_CTX *mem_ctx,
+		      const struct ldb_val *idx_val,
+		      struct ldb_val *key)
+{
+	struct ldb_context *ldb = ldb_module_get_ctx(module);
+	struct ldb_dn *dn;
+
+	if (ldb_kv->cache->GUID_index_attribute != NULL) {
+		return ldb_kv_guid_to_key(idx_val, key);
+	}
+
+	dn = ldb_dn_from_ldb_val(mem_ctx, ldb, idx_val);
+	if (dn == NULL) {
+		/*
+		 * LDB_ERR_INVALID_DN_SYNTAX would just be confusing
+		 * to the caller, as this in an invalid index value
+		 */
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+	/* form the key */
+	*key = ldb_kv_key_dn(mem_ctx, dn);
+	TALLOC_FREE(dn);
+	if (!key->data) {
+		return ldb_module_oom(module);
+	}
+	return LDB_SUCCESS;
+}
+
+/*
+  form a TDB_DATA for a record key
+  caller frees mem_ctx, which may or may not have the key
+  as a child.
+
+  note that the key for a record can depend on whether a
+  GUID index is in use, or the DN is used as the key
+*/
+struct ldb_val ldb_kv_key_msg(struct ldb_module *module,
+			TALLOC_CTX *mem_ctx,
+			const struct ldb_message *msg)
+{
+	void *data = ldb_module_get_private(module);
+	struct ldb_kv_private *ldb_kv =
+	    talloc_get_type(data, struct ldb_kv_private);
+	struct ldb_val key;
+	const struct ldb_val *guid_val;
+	int ret;
+
+	if (ldb_kv->cache->GUID_index_attribute == NULL) {
+		return ldb_kv_key_dn(mem_ctx, msg->dn);
+	}
+
+	if (ldb_dn_is_special(msg->dn)) {
+		return ldb_kv_key_dn(mem_ctx, msg->dn);
+	}
+
+	guid_val =
+	    ldb_msg_find_ldb_val(msg, ldb_kv->cache->GUID_index_attribute);
+	if (guid_val == NULL) {
+		ldb_asprintf_errstring(ldb_module_get_ctx(module),
+				       "Did not find GUID attribute %s "
+				       "in %s, required for TDB record "
+				       "key in " LDB_KV_IDXGUID " mode.",
+				       ldb_kv->cache->GUID_index_attribute,
+				       ldb_dn_get_linearized(msg->dn));
+		errno = EINVAL;
+		key.data = NULL;
+		key.length = 0;
+		return key;
+	}
+
+	/* In this case, allocate with talloc */
+	key.data = talloc_size(mem_ctx, LDB_KV_GUID_KEY_SIZE);
+	if (key.data == NULL) {
+		errno = ENOMEM;
+		key.data = NULL;
+		key.length = 0;
+		return key;
+	}
+	key.length = talloc_get_size(key.data);
+
+	ret = ldb_kv_guid_to_key(guid_val, &key);
+
+	if (ret != LDB_SUCCESS) {
+		errno = EINVAL;
+		key.data = NULL;
+		key.length = 0;
+		return key;
+	}
+	return key;
+}
+
+/*
+  check special dn's have valid attributes
+  currently only @ATTRIBUTES is checked
+*/
+static int ldb_kv_check_special_dn(struct ldb_module *module,
+				   const struct ldb_message *msg)
+{
+	struct ldb_context *ldb = ldb_module_get_ctx(module);
+	unsigned int i, j;
+
+	if (! ldb_dn_is_special(msg->dn) ||
+	    ! ldb_dn_check_special(msg->dn, LDB_KV_ATTRIBUTES)) {
+		return LDB_SUCCESS;
+	}
+
+	/* we have @ATTRIBUTES, let's check attributes are fine */
+	/* should we check that we deny multivalued attributes ? */
+	for (i = 0; i < msg->num_elements; i++) {
+		if (ldb_attr_cmp(msg->elements[i].name, "distinguishedName") == 0) continue;
+
+		for (j = 0; j < msg->elements[i].num_values; j++) {
+			if (ldb_kv_check_at_attributes_values(
+				&msg->elements[i].values[j]) != 0) {
+				ldb_set_errstring(ldb, "Invalid attribute value in an @ATTRIBUTES entry");
+				return LDB_ERR_INVALID_ATTRIBUTE_SYNTAX;
+			}
+		}
+	}
+
+	return LDB_SUCCESS;
+}
+
+/*
+ * Called after modifies and when starting a transaction. Checks target pack
+ * format version and current pack format version, which are set by cache_load,
+ * and repacks if necessary.
+ */
+static int ldb_kv_maybe_repack(struct ldb_kv_private *ldb_kv) {
+	/* Override option taken from ldb options */
+	if (ldb_kv->pack_format_override != 0) {
+		ldb_kv->target_pack_format_version =
+			ldb_kv->pack_format_override;
+	}
+
+	if (ldb_kv->pack_format_version !=
+	    ldb_kv->target_pack_format_version) {
+		int r;
+		struct ldb_context *ldb = ldb_module_get_ctx(ldb_kv->module);
+		r = ldb_kv_repack(ldb_kv->module);
+		if (r != LDB_SUCCESS) {
+			ldb_debug(ldb, LDB_DEBUG_ERROR,
+				  "Database repack failed.");
+		}
+		return r;
+	}
+
+	return LDB_SUCCESS;
+}
+
+/*
+  we've made a modification to a dn - possibly reindex and
+  update sequence number
+*/
+static int ldb_kv_modified(struct ldb_module *module, struct ldb_dn *dn)
+{
+	int ret = LDB_SUCCESS;
+	struct ldb_kv_private *ldb_kv = talloc_get_type(
+	    ldb_module_get_private(module), struct ldb_kv_private);
+
+	/* only allow modifies inside a transaction, otherwise the
+	 * ldb is unsafe */
+	if (ldb_kv->kv_ops->transaction_active(ldb_kv) == false) {
+		ldb_set_errstring(ldb_module_get_ctx(module), "ltdb modify without transaction");
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	if (ldb_dn_is_special(dn) &&
+	    (ldb_dn_check_special(dn, LDB_KV_INDEXLIST) ||
+	     ldb_dn_check_special(dn, LDB_KV_ATTRIBUTES)) )
+	{
+		if (ldb_kv->warn_reindex) {
+			ldb_debug(ldb_module_get_ctx(module),
+				  LDB_DEBUG_ERROR,
+				  "Reindexing %s due to modification on %s",
+				  ldb_kv->kv_ops->name(ldb_kv),
+				  ldb_dn_get_linearized(dn));
+		}
+		ret = ldb_kv_reindex(module);
+	}
+
+	/* If the modify was to a normal record, or any special except @BASEINFO, update the seq number */
+	if (ret == LDB_SUCCESS &&
+	    !(ldb_dn_is_special(dn) &&
+	      ldb_dn_check_special(dn, LDB_KV_BASEINFO)) ) {
+		ret = ldb_kv_increase_sequence_number(module);
+	}
+
+	/* If the modify was to @OPTIONS, reload the cache */
+	if (ret == LDB_SUCCESS &&
+	    ldb_dn_is_special(dn) &&
+	    (ldb_dn_check_special(dn, LDB_KV_OPTIONS)) ) {
+		ret = ldb_kv_cache_reload(module);
+	}
+
+	if (ret != LDB_SUCCESS) {
+		ldb_kv->reindex_failed = true;
+	}
+
+	return ret;
+}
+/*
+  store a record into the db
+*/
+int ldb_kv_store(struct ldb_module *module,
+		 const struct ldb_message *msg,
+		 int flgs)
+{
+	void *data = ldb_module_get_private(module);
+	struct ldb_kv_private *ldb_kv =
+	    talloc_get_type(data, struct ldb_kv_private);
+	struct ldb_val key;
+	struct ldb_val ldb_data;
+	int ret = LDB_SUCCESS;
+	TALLOC_CTX *key_ctx = talloc_new(module);
+
+	if (key_ctx == NULL) {
+		return ldb_module_oom(module);
+	}
+
+	if (ldb_kv->read_only) {
+		talloc_free(key_ctx);
+		return LDB_ERR_UNWILLING_TO_PERFORM;
+	}
+
+	key = ldb_kv_key_msg(module, key_ctx, msg);
+	if (key.data == NULL) {
+		TALLOC_FREE(key_ctx);
+		return LDB_ERR_OTHER;
+	}
+
+	ret = ldb_pack_data(ldb_module_get_ctx(module),
+			    msg, &ldb_data,
+			    ldb_kv->pack_format_version);
+	if (ret == -1) {
+		TALLOC_FREE(key_ctx);
+		return LDB_ERR_OTHER;
+	}
+
+	ret = ldb_kv->kv_ops->store(ldb_kv, key, ldb_data, flgs);
+	if (ret != 0) {
+		bool is_special = ldb_dn_is_special(msg->dn);
+		ret = ldb_kv->kv_ops->error(ldb_kv);
+
+		/*
+		 * LDB_ERR_ENTRY_ALREADY_EXISTS means the DN, not
+		 * the GUID, so re-map
+		 */
+		if (ret == LDB_ERR_ENTRY_ALREADY_EXISTS && !is_special &&
+		    ldb_kv->cache->GUID_index_attribute != NULL) {
+			ret = LDB_ERR_CONSTRAINT_VIOLATION;
+		}
+		goto done;
+	}
+
+done:
+	TALLOC_FREE(key_ctx);
+	talloc_free(ldb_data.data);
+
+	return ret;
+}
+
+
+/*
+  check if a attribute is a single valued, for a given element
+ */
+static bool ldb_kv_single_valued(const struct ldb_schema_attribute *a,
+				 struct ldb_message_element *el)
+{
+	if (!a) return false;
+	if (el != NULL) {
+		if (el->flags & LDB_FLAG_INTERNAL_FORCE_SINGLE_VALUE_CHECK) {
+			/* override from a ldb module, for example
+			   used for the description field, which is
+			   marked multi-valued in the schema but which
+			   should not actually accept multiple
+			   values */
+			return true;
+		}
+		if (el->flags & LDB_FLAG_INTERNAL_DISABLE_SINGLE_VALUE_CHECK) {
+			/* override from a ldb module, for example used for
+			   deleted linked attribute entries */
+			return false;
+		}
+	}
+	if (a->flags & LDB_ATTR_FLAG_SINGLE_VALUE) {
+		return true;
+	}
+	return false;
+}
+
+/*
+ * Starts a sub transaction if they are supported by the backend
+ * and the ldb connection has not been opened in batch mode.
+ */
+static int ldb_kv_sub_transaction_start(struct ldb_kv_private *ldb_kv)
+{
+	int ret = LDB_SUCCESS;
+
+	if (ldb_kv->batch_mode) {
+		return ret;
+	}
+
+	ret = ldb_kv->kv_ops->begin_nested_write(ldb_kv);
+	if (ret == LDB_SUCCESS) {
+		ret = ldb_kv_index_sub_transaction_start(ldb_kv);
+	}
+	return ret;
+}
+
+/*
+ * Commits a sub transaction if they are supported by the backend
+ * and the ldb connection has not been opened in batch mode.
+ */
+static int ldb_kv_sub_transaction_commit(struct ldb_kv_private *ldb_kv)
+{
+	int ret = LDB_SUCCESS;
+
+	if (ldb_kv->batch_mode) {
+		return ret;
+	}
+
+	ret = ldb_kv_index_sub_transaction_commit(ldb_kv);
+	if (ret != LDB_SUCCESS) {
+		return ret;
+	}
+	ret = ldb_kv->kv_ops->finish_nested_write(ldb_kv);
+	return ret;
+}
+
+/*
+ * Cancels a sub transaction if they are supported by the backend
+ * and the ldb connection has not been opened in batch mode.
+ */
+static int ldb_kv_sub_transaction_cancel(struct ldb_kv_private *ldb_kv)
+{
+	int ret = LDB_SUCCESS;
+
+	if (ldb_kv->batch_mode) {
+		return ret;
+	}
+
+	ret = ldb_kv_index_sub_transaction_cancel(ldb_kv);
+	if (ret != LDB_SUCCESS) {
+		struct ldb_context *ldb = ldb_module_get_ctx(ldb_kv->module);
+		/*
+		 * In the event of a failure we log the failure and continue
+		 * as we need to cancel the database transaction.
+		 */
+		ldb_debug(ldb,
+			  LDB_DEBUG_ERROR,
+			  __location__": ldb_kv_index_sub_transaction_cancel "
+			  "failed: %s",
+			  ldb_errstring(ldb));
+	}
+	ret = ldb_kv->kv_ops->abort_nested_write(ldb_kv);
+	return ret;
+}
+
+static int ldb_kv_add_internal(struct ldb_module *module,
+			       struct ldb_kv_private *ldb_kv,
+			       const struct ldb_message *msg,
+			       bool check_single_value)
+{
+	struct ldb_context *ldb = ldb_module_get_ctx(module);
+	int ret = LDB_SUCCESS;
+	unsigned int i;
+	bool valid_dn = false;
+
+	/* Check the new DN is reasonable */
+	valid_dn = ldb_dn_validate(msg->dn);
+	if (valid_dn == false) {
+		ldb_asprintf_errstring(ldb_module_get_ctx(module),
+				       "Invalid DN in ADD: %s",
+				       ldb_dn_get_linearized(msg->dn));
+		return LDB_ERR_INVALID_DN_SYNTAX;
+	}
+
+	for (i=0;i<msg->num_elements;i++) {
+		struct ldb_message_element *el = &msg->elements[i];
+		const struct ldb_schema_attribute *a = ldb_schema_attribute_by_name(ldb, el->name);
+
+		if (el->num_values == 0) {
+			ldb_asprintf_errstring(ldb, "attribute '%s' on '%s' specified, but with 0 values (illegal)",
+					       el->name, ldb_dn_get_linearized(msg->dn));
+			return LDB_ERR_CONSTRAINT_VIOLATION;
+		}
+		if (check_single_value && el->num_values > 1 &&
+		    ldb_kv_single_valued(a, el)) {
+			ldb_asprintf_errstring(ldb, "SINGLE-VALUE attribute %s on %s specified more than once",
+					       el->name, ldb_dn_get_linearized(msg->dn));
+			return LDB_ERR_CONSTRAINT_VIOLATION;
+		}
+
+		/* Do not check "@ATTRIBUTES" for duplicated values */
+		if (ldb_dn_is_special(msg->dn) &&
+		    ldb_dn_check_special(msg->dn, LDB_KV_ATTRIBUTES)) {
+			continue;
+		}
+
+		if (check_single_value &&
+		    !(el->flags &
+		      LDB_FLAG_INTERNAL_DISABLE_SINGLE_VALUE_CHECK)) {
+			struct ldb_val *duplicate = NULL;
+
+			ret = ldb_msg_find_duplicate_val(ldb, discard_const(msg),
+							 el, &duplicate, 0);
+			if (ret != LDB_SUCCESS) {
+				return ret;
+			}
+			if (duplicate != NULL) {
+				ldb_asprintf_errstring(
+					ldb,
+					"attribute '%s': value '%.*s' on '%s' "
+					"provided more than once in ADD object",
+					el->name,
+					(int)duplicate->length,
+					duplicate->data,
+					ldb_dn_get_linearized(msg->dn));
+				return LDB_ERR_ATTRIBUTE_OR_VALUE_EXISTS;
+			}
+		}
+	}
+
+	ret = ldb_kv_store(module, msg, TDB_INSERT);
+	if (ret != LDB_SUCCESS) {
+		/*
+		 * Try really hard to get the right error code for
+		 * a re-add situation, as this can matter!
+		 */
+		if (ret == LDB_ERR_CONSTRAINT_VIOLATION) {
+			int ret2;
+			struct ldb_dn *dn2 = NULL;
+			TALLOC_CTX *mem_ctx = talloc_new(module);
+			if (mem_ctx == NULL) {
+				return ldb_module_operr(module);
+			}
+			ret2 =
+			    ldb_kv_search_base(module, mem_ctx, msg->dn, &dn2);
+			TALLOC_FREE(mem_ctx);
+			if (ret2 == LDB_SUCCESS) {
+				ret = LDB_ERR_ENTRY_ALREADY_EXISTS;
+			}
+		}
+		if (ret == LDB_ERR_ENTRY_ALREADY_EXISTS) {
+			ldb_asprintf_errstring(ldb,
+					       "Entry %s already exists",
+					       ldb_dn_get_linearized(msg->dn));
+		}
+		return ret;
+	}
+
+	ret = ldb_kv_index_add_new(module, ldb_kv, msg);
+	if (ret != LDB_SUCCESS) {
+		/*
+		 * If we failed to index, delete the message again.
+		 *
+		 * This is particularly important for the GUID index
+		 * case, which will only fail for a duplicate DN
+		 * in the index add.
+		 *
+		 * Note that the caller may not cancel the transaction
+		 * and this means the above add might really show up!
+		 */
+		ldb_kv_delete_noindex(module, msg);
+		return ret;
+	}
+
+	ret = ldb_kv_modified(module, msg->dn);
+
+	/*
+	 * To allow testing of the error recovery code in ldb_kv_add
+	 * cmocka tests can define CMOCKA_UNIT_TEST_ADD_INTERNAL_FAIL
+	 * to inject failures at this point.
+	 */
+#ifdef CMOCKA_UNIT_TEST_ADD_INTERNAL_FAIL
+	CMOCKA_UNIT_TEST_ADD_INTERNAL_FAIL
+#endif
+
+	return ret;
+}
+
+/*
+  add a record to the database
+*/
+static int ldb_kv_add(struct ldb_kv_context *ctx)
+{
+	struct ldb_module *module = ctx->module;
+	struct ldb_request *req = ctx->req;
+	void *data = ldb_module_get_private(module);
+	struct ldb_kv_private *ldb_kv =
+	    talloc_get_type(data, struct ldb_kv_private);
+	int ret = LDB_SUCCESS;
+
+	if (ldb_kv->max_key_length != 0 &&
+	    ldb_kv->cache->GUID_index_attribute == NULL &&
+	    !ldb_dn_is_special(req->op.add.message->dn)) {
+		ldb_set_errstring(ldb_module_get_ctx(module),
+				  "Must operate ldb_mdb in GUID "
+				  "index mode, but " LDB_KV_IDXGUID " not set.");
+		return LDB_ERR_UNWILLING_TO_PERFORM;
+	}
+
+	ret = ldb_kv_check_special_dn(module, req->op.add.message);
+	if (ret != LDB_SUCCESS) {
+		return ret;
+	}
+
+	ldb_request_set_state(req, LDB_ASYNC_PENDING);
+
+	if (ldb_kv_cache_load(module) != 0) {
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	ret = ldb_kv_sub_transaction_start(ldb_kv);
+	if (ret != LDB_SUCCESS) {
+		return ret;
+	}
+	ret = ldb_kv_add_internal(module, ldb_kv, req->op.add.message, true);
+	if (ret != LDB_SUCCESS) {
+		int r = ldb_kv_sub_transaction_cancel(ldb_kv);
+		if (r != LDB_SUCCESS) {
+			ldb_debug(
+				ldb_module_get_ctx(module),
+				LDB_DEBUG_FATAL,
+				__location__
+				": Unable to roll back sub transaction");
+		}
+		ldb_kv->operation_failed = true;
+		return ret;
+	}
+	ret = ldb_kv_sub_transaction_commit(ldb_kv);
+
+	return ret;
+}
+
+/*
+  delete a record from the database, not updating indexes (used for deleting
+  index records)
+*/
+int ldb_kv_delete_noindex(struct ldb_module *module,
+			  const struct ldb_message *msg)
+{
+	void *data = ldb_module_get_private(module);
+	struct ldb_kv_private *ldb_kv =
+	    talloc_get_type(data, struct ldb_kv_private);
+	struct ldb_val key;
+	int ret;
+	TALLOC_CTX *tdb_key_ctx = talloc_new(module);
+
+	if (tdb_key_ctx == NULL) {
+		return ldb_module_oom(module);
+	}
+
+	if (ldb_kv->read_only) {
+		talloc_free(tdb_key_ctx);
+		return LDB_ERR_UNWILLING_TO_PERFORM;
+	}
+
+	key = ldb_kv_key_msg(module, tdb_key_ctx, msg);
+	if (!key.data) {
+		TALLOC_FREE(tdb_key_ctx);
+		return LDB_ERR_OTHER;
+	}
+
+	ret = ldb_kv->kv_ops->delete(ldb_kv, key);
+	TALLOC_FREE(tdb_key_ctx);
+
+	if (ret != 0) {
+		ret = ldb_kv->kv_ops->error(ldb_kv);
+	}
+
+	return ret;
+}
+
+static int ldb_kv_delete_internal(struct ldb_module *module, struct ldb_dn *dn)
+{
+	struct ldb_message *msg;
+	int ret = LDB_SUCCESS;
+
+	msg = ldb_msg_new(module);
+	if (msg == NULL) {
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	/* in case any attribute of the message was indexed, we need
+	   to fetch the old record */
+	ret = ldb_kv_search_dn1(module, dn, msg, 0);
+	if (ret != LDB_SUCCESS) {
+		/* not finding the old record is an error */
+		goto done;
+	}
+
+	ret = ldb_kv_delete_noindex(module, msg);
+	if (ret != LDB_SUCCESS) {
+		goto done;
+	}
+
+	/* remove any indexed attributes */
+	ret = ldb_kv_index_delete(module, msg);
+	if (ret != LDB_SUCCESS) {
+		goto done;
+	}
+
+	ret = ldb_kv_modified(module, dn);
+	if (ret != LDB_SUCCESS) {
+		goto done;
+	}
+
+done:
+	talloc_free(msg);
+	/*
+	 * To allow testing of the error recovery code in ldb_kv_delete
+	 * cmocka tests can define CMOCKA_UNIT_TEST_DELETE_INTERNAL_FAIL
+	 * to inject failures at this point.
+	 */
+#ifdef CMOCKA_UNIT_TEST_DELETE_INTERNAL_FAIL
+	CMOCKA_UNIT_TEST_DELETE_INTERNAL_FAIL
+#endif
+	return ret;
+}
+
+/*
+  delete a record from the database
+*/
+static int ldb_kv_delete(struct ldb_kv_context *ctx)
+{
+	struct ldb_module *module = ctx->module;
+	struct ldb_request *req = ctx->req;
+	void *data = ldb_module_get_private(module);
+	struct ldb_kv_private *ldb_kv =
+	    talloc_get_type(data, struct ldb_kv_private);
+	int ret = LDB_SUCCESS;
+
+	ldb_request_set_state(req, LDB_ASYNC_PENDING);
+
+	if (ldb_kv_cache_load(module) != 0) {
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	ret = ldb_kv_sub_transaction_start(ldb_kv);
+	if (ret != LDB_SUCCESS) {
+		return ret;
+	}
+	ret = ldb_kv_delete_internal(module, req->op.del.dn);
+	if (ret != LDB_SUCCESS) {
+		int r = ldb_kv_sub_transaction_cancel(ldb_kv);
+		if (r != LDB_SUCCESS) {
+			ldb_debug(
+				ldb_module_get_ctx(module),
+				LDB_DEBUG_FATAL,
+				__location__
+				": Unable to roll back sub transaction");
+		}
+		if (ret != LDB_ERR_NO_SUCH_OBJECT) {
+			ldb_kv->operation_failed = true;
+		}
+		return ret;
+	}
+	ret = ldb_kv_sub_transaction_commit(ldb_kv);
+
+	return ret;
+}
+
+/*
+  find an element by attribute name. At the moment this does a linear search,
+  it should be re-coded to use a binary search once all places that modify
+  records guarantee sorted order
+
+  return the index of the first matching element if found, otherwise -1
+*/
+static int ldb_kv_find_element(const struct ldb_message *msg, const char *name)
+{
+	unsigned int i;
+	for (i=0;i<msg->num_elements;i++) {
+		if (ldb_attr_cmp(msg->elements[i].name, name) == 0) {
+			return i;
+		}
+	}
+	return -1;
+}
+
+
+/*
+  add an element to an existing record. Assumes a elements array that we
+  can call re-alloc on, and assumes that we can reuse the data pointers from
+  the passed in additional values. Use with care!
+
+  returns 0 on success, -1 on failure (and sets errno)
+*/
+static int ldb_kv_msg_add_element(struct ldb_message *msg,
+				  struct ldb_message_element *el)
+{
+	struct ldb_message_element *e2;
+	unsigned int i;
+
+	if (el->num_values == 0) {
+		/* nothing to do here - we don't add empty elements */
+		return 0;
+	}
+
+	e2 = talloc_realloc(msg, msg->elements, struct ldb_message_element,
+			      msg->num_elements+1);
+	if (!e2) {
+		errno = ENOMEM;
+		return -1;
+	}
+
+	msg->elements = e2;
+
+	e2 = &msg->elements[msg->num_elements];
+
+	e2->name = el->name;
+	e2->flags = el->flags;
+	e2->values = talloc_array(msg->elements,
+				  struct ldb_val, el->num_values);
+	if (!e2->values) {
+		errno = ENOMEM;
+		return -1;
+	}
+	for (i=0;i<el->num_values;i++) {
+		e2->values[i] = el->values[i];
+	}
+	e2->num_values = el->num_values;
+
+	++msg->num_elements;
+
+	return 0;
+}
+
+/*
+  delete all elements having a specified attribute name
+*/
+static int ldb_kv_msg_delete_attribute(struct ldb_module *module,
+				       struct ldb_kv_private *ldb_kv,
+				       struct ldb_message *msg,
+				       const char *name)
+{
+	int ret;
+	struct ldb_message_element *el;
+	bool is_special = ldb_dn_is_special(msg->dn);
+
+	if (!is_special && ldb_kv->cache->GUID_index_attribute != NULL &&
+	    ldb_attr_cmp(name, ldb_kv->cache->GUID_index_attribute) == 0) {
+		struct ldb_context *ldb = ldb_module_get_ctx(module);
+		ldb_asprintf_errstring(ldb,
+				       "Must not modify GUID "
+				       "attribute %s (used as DB index)",
+				       ldb_kv->cache->GUID_index_attribute);
+		return LDB_ERR_CONSTRAINT_VIOLATION;
+	}
+
+	el = ldb_msg_find_element(msg, name);
+	if (el == NULL) {
+		return LDB_ERR_NO_SUCH_ATTRIBUTE;
+	}
+
+	ret = ldb_kv_index_del_element(module, ldb_kv, msg, el);
+	if (ret != LDB_SUCCESS) {
+		return ret;
+	}
+
+	talloc_free(el->values);
+	ldb_msg_remove_element(msg, el);
+	msg->elements = talloc_realloc(msg, msg->elements,
+				       struct ldb_message_element,
+				       msg->num_elements);
+	return LDB_SUCCESS;
+}
+
+/*
+  delete all elements matching an attribute name/value
+
+  return LDB Error on failure
+*/
+static int ldb_kv_msg_delete_element(struct ldb_module *module,
+				     struct ldb_kv_private *ldb_kv,
+				     struct ldb_message *msg,
+				     const char *name,
+				     const struct ldb_val *val)
+{
+	struct ldb_context *ldb = ldb_module_get_ctx(module);
+	unsigned int i;
+	int found, ret;
+	struct ldb_message_element *el;
+	const struct ldb_schema_attribute *a;
+
+	found = ldb_kv_find_element(msg, name);
+	if (found == -1) {
+		return LDB_ERR_NO_SUCH_ATTRIBUTE;
+	}
+
+	i = (unsigned int) found;
+	el = &(msg->elements[i]);
+
+	a = ldb_schema_attribute_by_name(ldb, el->name);
+
+	for (i=0;i<el->num_values;i++) {
+		bool matched;
+		if (a->syntax->operator_fn) {
+			ret = a->syntax->operator_fn(ldb, LDB_OP_EQUALITY, a,
+						     &el->values[i], val, &matched);
+			if (ret != LDB_SUCCESS) return ret;
+		} else {
+			matched = (a->syntax->comparison_fn(ldb, ldb,
+							    &el->values[i], val) == 0);
+		}
+		if (matched) {
+			if (el->num_values == 1) {
+				return ldb_kv_msg_delete_attribute(
+				    module, ldb_kv, msg, name);
+			}
+
+			ret =
+			    ldb_kv_index_del_value(module, ldb_kv, msg, el, i);
+			if (ret != LDB_SUCCESS) {
+				return ret;
+			}
+
+			ARRAY_DEL_ELEMENT(el->values, i, el->num_values);
+			el->num_values--;
+
+			/* per definition we find in a canonicalised message an
+			   attribute value only once. So we are finished here */
+			return LDB_SUCCESS;
+		}
+	}
+
+	/* Not found */
+	return LDB_ERR_NO_SUCH_ATTRIBUTE;
+}
+
+/*
+  modify a record - internal interface
+
+  yuck - this is O(n^2). Luckily n is usually small so we probably
+  get away with it, but if we ever have really large attribute lists
+  then we'll need to look at this again
+
+  'req' is optional, and is used to specify controls if supplied
+*/
+int ldb_kv_modify_internal(struct ldb_module *module,
+			   const struct ldb_message *msg,
+			   struct ldb_request *req)
+{
+	struct ldb_context *ldb = ldb_module_get_ctx(module);
+	void *data = ldb_module_get_private(module);
+	struct ldb_kv_private *ldb_kv =
+	    talloc_get_type(data, struct ldb_kv_private);
+	struct ldb_message *msg2;
+	unsigned int i, j;
+	int ret = LDB_SUCCESS, idx;
+	struct ldb_control *control_permissive = NULL;
+	TALLOC_CTX *mem_ctx = talloc_new(req);
+
+	if (mem_ctx == NULL) {
+		return ldb_module_oom(module);
+	}
+
+	if (req) {
+		control_permissive = ldb_request_get_control(req,
+					LDB_CONTROL_PERMISSIVE_MODIFY_OID);
+	}
+
+	msg2 = ldb_msg_new(mem_ctx);
+	if (msg2 == NULL) {
+		ret = LDB_ERR_OTHER;
+		goto done;
+	}
+
+	ret = ldb_kv_search_dn1(module, msg->dn, msg2, 0);
+	if (ret != LDB_SUCCESS) {
+		goto done;
+	}
+
+	for (i=0; i<msg->num_elements; i++) {
+		struct ldb_message_element *el = &msg->elements[i], *el2;
+		struct ldb_val *vals;
+		const struct ldb_schema_attribute *a = ldb_schema_attribute_by_name(ldb, el->name);
+		const char *dn;
+		uint32_t options = 0;
+		if (control_permissive != NULL) {
+			options |= LDB_MSG_FIND_COMMON_REMOVE_DUPLICATES;
+		}
+
+		switch (msg->elements[i].flags & LDB_FLAG_MOD_MASK) {
+		case LDB_FLAG_MOD_ADD:
+
+			if (el->num_values == 0) {
+				ldb_asprintf_errstring(ldb,
+						       "attribute '%s': attribute on '%s' specified, but with 0 values (illegal)",
+						       el->name, ldb_dn_get_linearized(msg2->dn));
+				ret = LDB_ERR_CONSTRAINT_VIOLATION;
+				goto done;
+			}
+
+			/* make a copy of the array so that a permissive
+			 * control can remove duplicates without changing the
+			 * original values, but do not copy data as we do not
+			 * need to keep it around once the operation is
+			 * finished */
+			if (control_permissive) {
+				el = talloc(msg2, struct ldb_message_element);
+				if (!el) {
+					ret = LDB_ERR_OTHER;
+					goto done;
+				}
+				*el = msg->elements[i];
+				el->values = talloc_array(el, struct ldb_val, el->num_values);
+				if (el->values == NULL) {
+					ret = LDB_ERR_OTHER;
+					goto done;
+				}
+				for (j = 0; j < el->num_values; j++) {
+					el->values[j] = msg->elements[i].values[j];
+				}
+			}
+
+			if (el->num_values > 1 && ldb_kv_single_valued(a, el)) {
+				ldb_asprintf_errstring(ldb, "SINGLE-VALUE attribute %s on %s specified more than once",
+						       el->name, ldb_dn_get_linearized(msg2->dn));
+				ret = LDB_ERR_ATTRIBUTE_OR_VALUE_EXISTS;
+				goto done;
+			}
+
+			/* Checks if element already exists */
+			idx = ldb_kv_find_element(msg2, el->name);
+			if (idx == -1) {
+				if (ldb_kv_msg_add_element(msg2, el) != 0) {
+					ret = LDB_ERR_OTHER;
+					goto done;
+				}
+				ret = ldb_kv_index_add_element(
+				    module, ldb_kv, msg2, el);
+				if (ret != LDB_SUCCESS) {
+					goto done;
+				}
+			} else {
+				j = (unsigned int) idx;
+				el2 = &(msg2->elements[j]);
+
+				/* We cannot add another value on a existing one
+				   if the attribute is single-valued */
+				if (ldb_kv_single_valued(a, el)) {
+					ldb_asprintf_errstring(ldb, "SINGLE-VALUE attribute %s on %s specified more than once",
+						               el->name, ldb_dn_get_linearized(msg2->dn));
+					ret = LDB_ERR_ATTRIBUTE_OR_VALUE_EXISTS;
+					goto done;
+				}
+
+				/* Check that values don't exist yet on multi-
+				   valued attributes or aren't provided twice */
+				if (!(el->flags &
+				      LDB_FLAG_INTERNAL_DISABLE_SINGLE_VALUE_CHECK)) {
+					struct ldb_val *duplicate = NULL;
+					ret = ldb_msg_find_common_values(ldb,
+									 msg2,
+									 el,
+									 el2,
+									 options);
+
+					if (ret ==
+					    LDB_ERR_ATTRIBUTE_OR_VALUE_EXISTS) {
+						ldb_asprintf_errstring(ldb,
+							"attribute '%s': value "
+							"#%u on '%s' already "
+							"exists", el->name, j,
+							ldb_dn_get_linearized(msg2->dn));
+						goto done;
+					} else if (ret != LDB_SUCCESS) {
+						goto done;
+					}
+
+					ret = ldb_msg_find_duplicate_val(
+						ldb, msg2, el, &duplicate, 0);
+					if (ret != LDB_SUCCESS) {
+						goto done;
+					}
+					if (duplicate != NULL) {
+						ldb_asprintf_errstring(
+							ldb,
+							"attribute '%s': value "
+							"'%.*s' on '%s' "
+							"provided more than "
+							"once in ADD",
+							el->name,
+							(int)duplicate->length,
+							duplicate->data,
+							ldb_dn_get_linearized(msg->dn));
+						ret = LDB_ERR_ATTRIBUTE_OR_VALUE_EXISTS;
+						goto done;
+					}
+				}
+
+				/* Now combine existing and new values to a new
+				   attribute record */
+				vals = talloc_realloc(msg2->elements,
+						      el2->values, struct ldb_val,
+						      el2->num_values + el->num_values);
+				if (vals == NULL) {
+					ldb_oom(ldb);
+					ret = LDB_ERR_OTHER;
+					goto done;
+				}
+
+				for (j=0; j<el->num_values; j++) {
+					vals[el2->num_values + j] =
+						ldb_val_dup(vals, &el->values[j]);
+				}
+
+				el2->values = vals;
+				el2->num_values += el->num_values;
+
+				ret = ldb_kv_index_add_element(
+				    module, ldb_kv, msg2, el);
+				if (ret != LDB_SUCCESS) {
+					goto done;
+				}
+			}
+
+			break;
+
+		case LDB_FLAG_MOD_REPLACE:
+
+			if (el->num_values > 1 && ldb_kv_single_valued(a, el)) {
+				ldb_asprintf_errstring(ldb, "SINGLE-VALUE attribute %s on %s specified more than once",
+						       el->name, ldb_dn_get_linearized(msg2->dn));
+				ret = LDB_ERR_ATTRIBUTE_OR_VALUE_EXISTS;
+				goto done;
+			}
+
+			/*
+			 * We don't need to check this if we have been
+			 * pre-screened by the repl_meta_data module
+			 * in Samba, or someone else who can claim to
+			 * know what they are doing.
+			 */
+			if (!(el->flags & LDB_FLAG_INTERNAL_DISABLE_SINGLE_VALUE_CHECK)) {
+				struct ldb_val *duplicate = NULL;
+
+				ret = ldb_msg_find_duplicate_val(ldb, msg2, el,
+								 &duplicate, 0);
+				if (ret != LDB_SUCCESS) {
+					goto done;
+				}
+				if (duplicate != NULL) {
+					ldb_asprintf_errstring(
+						ldb,
+						"attribute '%s': value '%.*s' "
+						"on '%s' provided more than "
+						"once in REPLACE",
+						el->name,
+						(int)duplicate->length,
+						duplicate->data,
+						ldb_dn_get_linearized(msg2->dn));
+					ret = LDB_ERR_ATTRIBUTE_OR_VALUE_EXISTS;
+					goto done;
+				}
+			}
+
+			/* Checks if element already exists */
+			idx = ldb_kv_find_element(msg2, el->name);
+			if (idx != -1) {
+				j = (unsigned int) idx;
+				el2 = &(msg2->elements[j]);
+
+				/* we consider two elements to be
+				 * equal only if the order
+				 * matches. This allows dbcheck to
+				 * fix the ordering on attributes
+				 * where order matters, such as
+				 * objectClass
+				 */
+				if (ldb_msg_element_equal_ordered(el, el2)) {
+					continue;
+				}
+
+				/* Delete the attribute if it exists in the DB */
+				if (ldb_kv_msg_delete_attribute(
+					module, ldb_kv, msg2, el->name) != 0) {
+					ret = LDB_ERR_OTHER;
+					goto done;
+				}
+			}
+
+			/* Recreate it with the new values */
+			if (ldb_kv_msg_add_element(msg2, el) != 0) {
+				ret = LDB_ERR_OTHER;
+				goto done;
+			}
+
+			ret =
+			    ldb_kv_index_add_element(module, ldb_kv, msg2, el);
+			if (ret != LDB_SUCCESS) {
+				goto done;
+			}
+
+			break;
+
+		case LDB_FLAG_MOD_DELETE:
+			dn = ldb_dn_get_linearized(msg2->dn);
+			if (dn == NULL) {
+				ret = LDB_ERR_OTHER;
+				goto done;
+			}
+
+			if (msg->elements[i].num_values == 0) {
+				/* Delete the whole attribute */
+				ret = ldb_kv_msg_delete_attribute(
+				    module,
+				    ldb_kv,
+				    msg2,
+				    msg->elements[i].name);
+				if (ret == LDB_ERR_NO_SUCH_ATTRIBUTE) {
+					if (control_permissive) {
+						ret = LDB_SUCCESS;
+					} else {
+						ldb_asprintf_errstring(ldb,
+								       "attribute '%s': no such attribute for delete on '%s'",
+								       msg->elements[i].name, dn);
+					}
+				}
+				if (ret != LDB_SUCCESS) {
+					goto done;
+				}
+			} else {
+				/* Delete specified values from an attribute */
+				for (j=0; j < msg->elements[i].num_values; j++) {
+					ret = ldb_kv_msg_delete_element(
+					    module,
+					    ldb_kv,
+					    msg2,
+					    msg->elements[i].name,
+					    &msg->elements[i].values[j]);
+					if (ret == LDB_ERR_NO_SUCH_ATTRIBUTE &&
+					    control_permissive) {
+						ret = LDB_SUCCESS;
+					} else if (ret == LDB_ERR_NO_SUCH_ATTRIBUTE) {
+						ldb_asprintf_errstring(ldb,
+								       "attribute '%s': no matching attribute value while deleting attribute on '%s'",
+								       msg->elements[i].name, dn);
+					}
+					if (ret != LDB_SUCCESS) {
+						goto done;
+					}
+				}
+			}
+			break;
+		default:
+			ldb_asprintf_errstring(ldb,
+					       "attribute '%s': invalid modify flags on '%s': 0x%x",
+					       msg->elements[i].name, ldb_dn_get_linearized(msg->dn),
+					       msg->elements[i].flags & LDB_FLAG_MOD_MASK);
+			ret = LDB_ERR_PROTOCOL_ERROR;
+			goto done;
+		}
+	}
+
+	ret = ldb_kv_store(module, msg2, TDB_MODIFY);
+	if (ret != LDB_SUCCESS) {
+		goto done;
+	}
+
+	ret = ldb_kv_modified(module, msg2->dn);
+	if (ret != LDB_SUCCESS) {
+		goto done;
+	}
+
+done:
+	TALLOC_FREE(mem_ctx);
+	/*
+	 * To allow testing of the error recovery code in ldb_kv_modify
+	 * cmocka tests can define CMOCKA_UNIT_TEST_MODIFY_INTERNAL_FAIL
+	 * to inject failures at this point.
+	 */
+#ifdef CMOCKA_UNIT_TEST_MODIFY_INTERNAL_FAIL
+	CMOCKA_UNIT_TEST_MODIFY_INTERNAL_FAIL
+#endif
+	return ret;
+}
+
+/*
+  modify a record
+*/
+static int ldb_kv_modify(struct ldb_kv_context *ctx)
+{
+	struct ldb_module *module = ctx->module;
+	struct ldb_request *req = ctx->req;
+	void *data = ldb_module_get_private(module);
+	struct ldb_kv_private *ldb_kv =
+	    talloc_get_type(data, struct ldb_kv_private);
+	int ret = LDB_SUCCESS;
+
+	ret = ldb_kv_check_special_dn(module, req->op.mod.message);
+	if (ret != LDB_SUCCESS) {
+		return ret;
+	}
+
+	ldb_request_set_state(req, LDB_ASYNC_PENDING);
+
+	if (ldb_kv_cache_load(module) != 0) {
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	ret = ldb_kv_sub_transaction_start(ldb_kv);
+	if (ret != LDB_SUCCESS) {
+		return ret;
+	}
+	ret = ldb_kv_modify_internal(module, req->op.mod.message, req);
+	if (ret != LDB_SUCCESS) {
+		int r = ldb_kv_sub_transaction_cancel(ldb_kv);
+		if (r != LDB_SUCCESS) {
+			ldb_debug(
+				ldb_module_get_ctx(module),
+				LDB_DEBUG_FATAL,
+				__location__
+				": Unable to roll back sub transaction");
+		}
+		if (ret != LDB_ERR_NO_SUCH_OBJECT) {
+			ldb_kv->operation_failed = true;
+		}
+		return ret;
+	}
+	ret = ldb_kv_sub_transaction_commit(ldb_kv);
+
+
+	return ret;
+}
+
+static int ldb_kv_rename_internal(struct ldb_module *module,
+			   struct ldb_request *req,
+			   struct ldb_message *msg)
+{
+	void *data = ldb_module_get_private(module);
+	struct ldb_kv_private *ldb_kv =
+	    talloc_get_type(data, struct ldb_kv_private);
+	int ret = LDB_SUCCESS;
+
+	/* Always delete first then add, to avoid conflicts with
+	 * unique indexes. We rely on the transaction to make this
+	 * atomic
+	 */
+	ret = ldb_kv_delete_internal(module, msg->dn);
+	if (ret != LDB_SUCCESS) {
+		return ret;
+	}
+
+	msg->dn = ldb_dn_copy(msg, req->op.rename.newdn);
+	if (msg->dn == NULL) {
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	/* We don't check single value as we can have more than 1 with
+	 * deleted attributes. We could go through all elements but that's
+	 * maybe not the most efficient way
+	 */
+	ret = ldb_kv_add_internal(module, ldb_kv, msg, false);
+
+	/*
+	 * To allow testing of the error recovery code in ldb_kv_rename
+	 * cmocka tests can define CMOCKA_UNIT_TEST_RENAME_INTERNAL_FAIL
+	 * to inject failures at this point.
+	 */
+#ifdef CMOCKA_UNIT_TEST_RENAME_INTERNAL_FAIL
+	CMOCKA_UNIT_TEST_RENAME_INTERNAL_FAIL
+#endif
+	return ret;
+}
+
+/*
+  rename a record
+*/
+static int ldb_kv_rename(struct ldb_kv_context *ctx)
+{
+	struct ldb_module *module = ctx->module;
+	void *data = ldb_module_get_private(module);
+	struct ldb_kv_private *ldb_kv =
+	    talloc_get_type(data, struct ldb_kv_private);
+	struct ldb_request *req = ctx->req;
+	struct ldb_message *msg;
+	int ret = LDB_SUCCESS;
+	struct ldb_val  key, key_old;
+	struct ldb_dn *db_dn;
+	bool valid_dn = false;
+
+	ldb_request_set_state(req, LDB_ASYNC_PENDING);
+
+	if (ldb_kv_cache_load(ctx->module) != 0) {
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	msg = ldb_msg_new(ctx);
+	if (msg == NULL) {
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	/* Check the new DN is reasonable */
+	valid_dn = ldb_dn_validate(req->op.rename.newdn);
+	if (valid_dn == false) {
+		ldb_asprintf_errstring(ldb_module_get_ctx(module),
+				       "Invalid New DN: %s",
+				       ldb_dn_get_linearized(req->op.rename.newdn));
+		return LDB_ERR_INVALID_DN_SYNTAX;
+	}
+
+	/* we need to fetch the old record to re-add under the new name */
+	ret = ldb_kv_search_dn1(module, req->op.rename.olddn, msg, 0);
+	if (ret == LDB_ERR_INVALID_DN_SYNTAX) {
+		ldb_asprintf_errstring(ldb_module_get_ctx(module),
+				       "Invalid Old DN: %s",
+				       ldb_dn_get_linearized(req->op.rename.newdn));
+		return ret;
+	} else if (ret != LDB_SUCCESS) {
+		/* not finding the old record is an error */
+		return ret;
+	}
+
+	/* We need to, before changing the DB, check if the new DN
+	 * exists, so we can return this error to the caller with an
+	 * unmodified DB
+	 *
+	 * Even in GUID index mode we use ltdb_key_dn() as we are
+	 * trying to figure out if this is just a case rename
+	 */
+	key = ldb_kv_key_dn(msg, req->op.rename.newdn);
+	if (!key.data) {
+		talloc_free(msg);
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	key_old = ldb_kv_key_dn(msg, req->op.rename.olddn);
+	if (!key_old.data) {
+		talloc_free(msg);
+		talloc_free(key.data);
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	/*
+	 * Only declare a conflict if the new DN already exists,
+	 * and it isn't a case change on the old DN
+	 */
+	if (key_old.length != key.length
+	    || memcmp(key.data, key_old.data, key.length) != 0) {
+		ret = ldb_kv_search_base(
+		    module, msg, req->op.rename.newdn, &db_dn);
+		if (ret == LDB_SUCCESS) {
+			ret = LDB_ERR_ENTRY_ALREADY_EXISTS;
+		} else if (ret == LDB_ERR_NO_SUCH_OBJECT) {
+			ret = LDB_SUCCESS;
+		}
+	}
+
+	/* finding the new record already in the DB is an error */
+
+	if (ret == LDB_ERR_ENTRY_ALREADY_EXISTS) {
+		ldb_asprintf_errstring(ldb_module_get_ctx(module),
+				       "Entry %s already exists",
+				       ldb_dn_get_linearized(req->op.rename.newdn));
+	}
+	if (ret != LDB_SUCCESS) {
+		talloc_free(key_old.data);
+		talloc_free(key.data);
+		talloc_free(msg);
+		return ret;
+	}
+
+	talloc_free(key_old.data);
+	talloc_free(key.data);
+
+
+	ret = ldb_kv_sub_transaction_start(ldb_kv);
+	if (ret != LDB_SUCCESS) {
+		talloc_free(msg);
+		return ret;
+	}
+	ret = ldb_kv_rename_internal(module, req, msg);
+	if (ret != LDB_SUCCESS) {
+		int r = ldb_kv_sub_transaction_cancel(ldb_kv);
+		if (r != LDB_SUCCESS) {
+			ldb_debug(
+				ldb_module_get_ctx(module),
+				LDB_DEBUG_FATAL,
+				__location__
+				": Unable to roll back sub transaction");
+		}
+		talloc_free(msg);
+		ldb_kv->operation_failed = true;
+		return ret;
+	}
+	ret = ldb_kv_sub_transaction_commit(ldb_kv);
+	talloc_free(msg);
+
+	return ret;
+}
+
+static int ldb_kv_start_trans(struct ldb_module *module)
+{
+	void *data = ldb_module_get_private(module);
+	struct ldb_kv_private *ldb_kv =
+	    talloc_get_type(data, struct ldb_kv_private);
+
+	pid_t pid = getpid();
+
+	if (ldb_kv->pid != pid) {
+		ldb_asprintf_errstring(ldb_module_get_ctx(ldb_kv->module),
+				       __location__
+				       ": Reusing ldb opened by pid %d in "
+				       "process %d\n",
+				       ldb_kv->pid,
+				       pid);
+		return LDB_ERR_PROTOCOL_ERROR;
+	}
+
+	/* Do not take out the transaction lock on a read-only DB */
+	if (ldb_kv->read_only) {
+		return LDB_ERR_UNWILLING_TO_PERFORM;
+	}
+
+	if (ldb_kv->kv_ops->begin_write(ldb_kv) != 0) {
+		return ldb_kv->kv_ops->error(ldb_kv);
+	}
+
+	ldb_kv_index_transaction_start(
+		module,
+		ldb_kv->index_transaction_cache_size);
+
+	ldb_kv->reindex_failed = false;
+	ldb_kv->operation_failed = false;
+
+	return LDB_SUCCESS;
+}
+
+/*
+ * Forward declaration to allow prepare_commit to in fact abort the
+ * transaction
+ */
+static int ldb_kv_del_trans(struct ldb_module *module);
+
+static int ldb_kv_prepare_commit(struct ldb_module *module)
+{
+	int ret;
+	void *data = ldb_module_get_private(module);
+	struct ldb_kv_private *ldb_kv =
+	    talloc_get_type(data, struct ldb_kv_private);
+	pid_t pid = getpid();
+
+	if (ldb_kv->pid != pid) {
+		ldb_asprintf_errstring(ldb_module_get_ctx(module),
+				       __location__
+				       ": Reusing ldb opened by pid %d in "
+				       "process %d\n",
+				       ldb_kv->pid,
+				       pid);
+		return LDB_ERR_PROTOCOL_ERROR;
+	}
+
+	if (!ldb_kv->kv_ops->transaction_active(ldb_kv)) {
+		ldb_set_errstring(ldb_module_get_ctx(module),
+				  "ltdb_prepare_commit() called "
+				  "without transaction active");
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	/*
+	 * Check if the last re-index failed.
+	 *
+	 * This can happen if for example a duplicate value was marked
+	 * unique.  We must not write a partial re-index into the DB.
+	 */
+	if (ldb_kv->reindex_failed) {
+		/*
+		 * We must instead abort the transaction so we get the
+		 * old values and old index back
+		 */
+		ldb_kv_del_trans(module);
+		ldb_set_errstring(ldb_module_get_ctx(module),
+				  "Failure during re-index, so "
+				  "transaction must be aborted.");
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	ret = ldb_kv_index_transaction_commit(module);
+	if (ret != LDB_SUCCESS) {
+		ldb_kv->kv_ops->abort_write(ldb_kv);
+		return ret;
+	}
+
+	/*
+	 * If GUID indexing was toggled in this transaction, we repack at
+	 * format version 2 if GUID indexing was enabled, or version 1 if
+	 * it was disabled.
+	 */
+	ret = ldb_kv_maybe_repack(ldb_kv);
+	if (ret != LDB_SUCCESS) {
+		ldb_kv_del_trans(module);
+		ldb_set_errstring(ldb_module_get_ctx(module),
+				  "Failure during re-pack, so "
+				  "transaction must be aborted.");
+		return ret;
+	}
+
+	if (ldb_kv->kv_ops->prepare_write(ldb_kv) != 0) {
+		ret = ldb_kv->kv_ops->error(ldb_kv);
+		ldb_debug_set(ldb_module_get_ctx(module),
+			      LDB_DEBUG_FATAL,
+			      "Failure during "
+			      "prepare_write): %s -> %s",
+			      ldb_kv->kv_ops->errorstr(ldb_kv),
+			      ldb_strerror(ret));
+		return ret;
+	}
+
+	ldb_kv->prepared_commit = true;
+
+	return LDB_SUCCESS;
+}
+
+static int ldb_kv_end_trans(struct ldb_module *module)
+{
+	int ret;
+	void *data = ldb_module_get_private(module);
+	struct ldb_kv_private *ldb_kv =
+	    talloc_get_type(data, struct ldb_kv_private);
+
+	/*
+	 * If in batch mode and there has been an operation failure
+	 * rollback the transaction rather than committing it to avoid
+	 * any possible corruption
+	 */
+	if (ldb_kv->batch_mode && ldb_kv->operation_failed) {
+		ret = ldb_kv_del_trans( module);
+		if (ret != LDB_SUCCESS) {
+			ldb_debug_set(ldb_module_get_ctx(module),
+				      LDB_DEBUG_FATAL,
+				      "An operation failed during a batch mode "
+				      "transaction. The transaction could not"
+				      "be rolled back, ldb_kv_del_trans "
+				      "returned (%s, %s)",
+				      ldb_kv->kv_ops->errorstr(ldb_kv),
+				      ldb_strerror(ret));
+		} else {
+			ldb_debug_set(ldb_module_get_ctx(module),
+				      LDB_DEBUG_FATAL,
+				      "An operation failed during a batch mode "
+				      "transaction, the transaction was "
+				      "rolled back");
+		}
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	if (!ldb_kv->prepared_commit) {
+		ret = ldb_kv_prepare_commit(module);
+		if (ret != LDB_SUCCESS) {
+			return ret;
+		}
+	}
+
+	ldb_kv->prepared_commit = false;
+
+	if (ldb_kv->kv_ops->finish_write(ldb_kv) != 0) {
+		ret = ldb_kv->kv_ops->error(ldb_kv);
+		ldb_asprintf_errstring(
+		    ldb_module_get_ctx(module),
+		    "Failure during tdb_transaction_commit(): %s -> %s",
+		    ldb_kv->kv_ops->errorstr(ldb_kv),
+		    ldb_strerror(ret));
+		return ret;
+	}
+
+	return LDB_SUCCESS;
+}
+
+static int ldb_kv_del_trans(struct ldb_module *module)
+{
+	void *data = ldb_module_get_private(module);
+	struct ldb_kv_private *ldb_kv =
+	    talloc_get_type(data, struct ldb_kv_private);
+
+	if (ldb_kv_index_transaction_cancel(module) != 0) {
+		ldb_kv->kv_ops->abort_write(ldb_kv);
+		return ldb_kv->kv_ops->error(ldb_kv);
+	}
+
+	ldb_kv->kv_ops->abort_write(ldb_kv);
+	return LDB_SUCCESS;
+}
+
+/*
+  return sequenceNumber from @BASEINFO
+*/
+static int ldb_kv_sequence_number(struct ldb_kv_context *ctx,
+				  struct ldb_extended **ext)
+{
+	struct ldb_context *ldb;
+	struct ldb_module *module = ctx->module;
+	struct ldb_request *req = ctx->req;
+	void *data = ldb_module_get_private(module);
+	struct ldb_kv_private *ldb_kv =
+	    talloc_get_type(data, struct ldb_kv_private);
+	TALLOC_CTX *tmp_ctx = NULL;
+	struct ldb_seqnum_request *seq;
+	struct ldb_seqnum_result *res;
+	struct ldb_message *msg = NULL;
+	struct ldb_dn *dn;
+	const char *date;
+	int ret = LDB_SUCCESS;
+
+	ldb = ldb_module_get_ctx(module);
+
+	seq = talloc_get_type(req->op.extended.data,
+				struct ldb_seqnum_request);
+	if (seq == NULL) {
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	ldb_request_set_state(req, LDB_ASYNC_PENDING);
+
+	if (ldb_kv->kv_ops->lock_read(module) != 0) {
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	res = talloc_zero(req, struct ldb_seqnum_result);
+	if (res == NULL) {
+		ret = LDB_ERR_OPERATIONS_ERROR;
+		goto done;
+	}
+
+	tmp_ctx = talloc_new(req);
+	if (tmp_ctx == NULL) {
+		ret = LDB_ERR_OPERATIONS_ERROR;
+		goto done;
+	}
+
+	dn = ldb_dn_new(tmp_ctx, ldb, LDB_KV_BASEINFO);
+	if (dn == NULL) {
+		ret = LDB_ERR_OPERATIONS_ERROR;
+		goto done;
+	}
+
+	msg = ldb_msg_new(tmp_ctx);
+	if (msg == NULL) {
+		ret = LDB_ERR_OPERATIONS_ERROR;
+		goto done;
+	}
+
+	ret = ldb_kv_search_dn1(module, dn, msg, 0);
+	if (ret != LDB_SUCCESS) {
+		goto done;
+	}
+
+	switch (seq->type) {
+	case LDB_SEQ_HIGHEST_SEQ:
+		res->seq_num = ldb_msg_find_attr_as_uint64(msg, LDB_KV_SEQUENCE_NUMBER, 0);
+		break;
+	case LDB_SEQ_NEXT:
+		res->seq_num = ldb_msg_find_attr_as_uint64(msg, LDB_KV_SEQUENCE_NUMBER, 0);
+		res->seq_num++;
+		break;
+	case LDB_SEQ_HIGHEST_TIMESTAMP:
+		date = ldb_msg_find_attr_as_string(msg, LDB_KV_MOD_TIMESTAMP, NULL);
+		if (date) {
+			res->seq_num = ldb_string_to_time(date);
+		} else {
+			res->seq_num = 0;
+			/* zero is as good as anything when we don't know */
+		}
+		break;
+	}
+
+	*ext = talloc_zero(req, struct ldb_extended);
+	if (*ext == NULL) {
+		ret = LDB_ERR_OPERATIONS_ERROR;
+		goto done;
+	}
+	(*ext)->oid = LDB_EXTENDED_SEQUENCE_NUMBER;
+	(*ext)->data = talloc_steal(*ext, res);
+
+done:
+	talloc_free(tmp_ctx);
+
+	ldb_kv->kv_ops->unlock_read(module);
+	return ret;
+}
+
+static void ldb_kv_request_done(struct ldb_kv_context *ctx, int error)
+{
+	struct ldb_context *ldb;
+	struct ldb_request *req;
+	struct ldb_reply *ares;
+
+	ldb = ldb_module_get_ctx(ctx->module);
+	req = ctx->req;
+
+	/* if we already returned an error just return */
+	if (ldb_request_get_status(req) != LDB_SUCCESS) {
+		return;
+	}
+
+	ares = talloc_zero(req, struct ldb_reply);
+	if (!ares) {
+		ldb_oom(ldb);
+		req->callback(req, NULL);
+		return;
+	}
+	ares->type = LDB_REPLY_DONE;
+	ares->error = error;
+
+	req->callback(req, ares);
+}
+
+static void ldb_kv_timeout(_UNUSED_ struct tevent_context *ev,
+			   _UNUSED_ struct tevent_timer *te,
+			   _UNUSED_ struct timeval t,
+			   void *private_data)
+{
+	struct ldb_kv_context *ctx;
+	ctx = talloc_get_type(private_data, struct ldb_kv_context);
+
+	if (!ctx->request_terminated) {
+		/* request is done now */
+		ldb_kv_request_done(ctx, LDB_ERR_TIME_LIMIT_EXCEEDED);
+	}
+
+	if (ctx->spy) {
+		/* neutralize the spy */
+		ctx->spy->ctx = NULL;
+		ctx->spy = NULL;
+	}
+	talloc_free(ctx);
+}
+
+static void ldb_kv_request_extended_done(struct ldb_kv_context *ctx,
+					 struct ldb_extended *ext,
+					 int error)
+{
+	struct ldb_context *ldb;
+	struct ldb_request *req;
+	struct ldb_reply *ares;
+
+	ldb = ldb_module_get_ctx(ctx->module);
+	req = ctx->req;
+
+	/* if we already returned an error just return */
+	if (ldb_request_get_status(req) != LDB_SUCCESS) {
+		return;
+	}
+
+	ares = talloc_zero(req, struct ldb_reply);
+	if (!ares) {
+		ldb_oom(ldb);
+		req->callback(req, NULL);
+		return;
+	}
+	ares->type = LDB_REPLY_DONE;
+	ares->response = ext;
+	ares->error = error;
+
+	req->callback(req, ares);
+}
+
+static void ldb_kv_handle_extended(struct ldb_kv_context *ctx)
+{
+	struct ldb_extended *ext = NULL;
+	int ret;
+
+	if (strcmp(ctx->req->op.extended.oid,
+		   LDB_EXTENDED_SEQUENCE_NUMBER) == 0) {
+		/* get sequence number */
+		ret = ldb_kv_sequence_number(ctx, &ext);
+	} else {
+		/* not recognized */
+		ret = LDB_ERR_UNSUPPORTED_CRITICAL_EXTENSION;
+	}
+
+	ldb_kv_request_extended_done(ctx, ext, ret);
+}
+
+static void ldb_kv_callback(struct tevent_context *ev,
+			    struct tevent_timer *te,
+			    struct timeval t,
+			    void *private_data)
+{
+	struct ldb_kv_context *ctx;
+	int ret;
+
+	ctx = talloc_get_type(private_data, struct ldb_kv_context);
+
+	if (ctx->request_terminated) {
+		goto done;
+	}
+
+	switch (ctx->req->operation) {
+	case LDB_SEARCH:
+		ret = ldb_kv_search(ctx);
+		break;
+	case LDB_ADD:
+		ret = ldb_kv_add(ctx);
+		break;
+	case LDB_MODIFY:
+		ret = ldb_kv_modify(ctx);
+		break;
+	case LDB_DELETE:
+		ret = ldb_kv_delete(ctx);
+		break;
+	case LDB_RENAME:
+		ret = ldb_kv_rename(ctx);
+		break;
+	case LDB_EXTENDED:
+		ldb_kv_handle_extended(ctx);
+		goto done;
+	default:
+		/* no other op supported */
+		ret = LDB_ERR_PROTOCOL_ERROR;
+	}
+
+	if (!ctx->request_terminated) {
+		/* request is done now */
+		ldb_kv_request_done(ctx, ret);
+	}
+
+done:
+	if (ctx->spy) {
+		/* neutralize the spy */
+		ctx->spy->ctx = NULL;
+		ctx->spy = NULL;
+	}
+	talloc_free(ctx);
+}
+
+static int ldb_kv_request_destructor(void *ptr)
+{
+	struct ldb_kv_req_spy *spy =
+	    talloc_get_type(ptr, struct ldb_kv_req_spy);
+
+	if (spy->ctx != NULL) {
+		spy->ctx->spy = NULL;
+		spy->ctx->request_terminated = true;
+		spy->ctx = NULL;
+	}
+
+	return 0;
+}
+
+static int ldb_kv_handle_request(struct ldb_module *module,
+				 struct ldb_request *req)
+{
+	struct ldb_control *control_permissive;
+	struct ldb_context *ldb;
+	struct tevent_context *ev;
+	struct ldb_kv_context *ac;
+	struct tevent_timer *te;
+	struct timeval tv;
+	unsigned int i;
+
+	ldb = ldb_module_get_ctx(module);
+
+	control_permissive = ldb_request_get_control(req,
+					LDB_CONTROL_PERMISSIVE_MODIFY_OID);
+
+	for (i = 0; req->controls && req->controls[i]; i++) {
+		if (req->controls[i]->critical &&
+		    req->controls[i] != control_permissive) {
+			ldb_asprintf_errstring(ldb, "Unsupported critical extension %s",
+					       req->controls[i]->oid);
+			return LDB_ERR_UNSUPPORTED_CRITICAL_EXTENSION;
+		}
+	}
+
+	if (req->starttime == 0 || req->timeout == 0) {
+		ldb_set_errstring(ldb, "Invalid timeout settings");
+		return LDB_ERR_TIME_LIMIT_EXCEEDED;
+	}
+
+	ev = ldb_handle_get_event_context(req->handle);
+
+	ac = talloc_zero(ldb, struct ldb_kv_context);
+	if (ac == NULL) {
+		ldb_oom(ldb);
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	ac->module = module;
+	ac->req = req;
+
+	tv.tv_sec = 0;
+	tv.tv_usec = 0;
+	te = tevent_add_timer(ev, ac, tv, ldb_kv_callback, ac);
+	if (NULL == te) {
+		talloc_free(ac);
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	if (req->timeout > 0) {
+		tv.tv_sec = req->starttime + req->timeout;
+		tv.tv_usec = 0;
+		ac->timeout_event =
+		    tevent_add_timer(ev, ac, tv, ldb_kv_timeout, ac);
+		if (NULL == ac->timeout_event) {
+			talloc_free(ac);
+			return LDB_ERR_OPERATIONS_ERROR;
+		}
+	}
+
+	ac->timeout_timeval = tv;
+
+	/* set a spy so that we do not try to use the request context
+	 * if it is freed before ltdb_callback fires */
+	ac->spy = talloc(req, struct ldb_kv_req_spy);
+	if (NULL == ac->spy) {
+		talloc_free(ac);
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+	ac->spy->ctx = ac;
+
+	talloc_set_destructor((TALLOC_CTX *)ac->spy, ldb_kv_request_destructor);
+
+	return LDB_SUCCESS;
+}
+
+static int ldb_kv_init_rootdse(struct ldb_module *module)
+{
+	/* ignore errors on this - we expect it for non-sam databases */
+	ldb_mod_register_control(module, LDB_CONTROL_PERMISSIVE_MODIFY_OID);
+
+	/* there can be no module beyond the backend, just return */
+	return LDB_SUCCESS;
+}
+
+static int ldb_kv_lock_read(struct ldb_module *module)
+{
+	void *data = ldb_module_get_private(module);
+	struct ldb_kv_private *ldb_kv =
+	    talloc_get_type(data, struct ldb_kv_private);
+	return ldb_kv->kv_ops->lock_read(module);
+}
+
+static int ldb_kv_unlock_read(struct ldb_module *module)
+{
+	void *data = ldb_module_get_private(module);
+	struct ldb_kv_private *ldb_kv =
+	    talloc_get_type(data, struct ldb_kv_private);
+	return ldb_kv->kv_ops->unlock_read(module);
+}
+
+static const struct ldb_module_ops ldb_kv_ops = {
+    .name = "tdb",
+    .init_context = ldb_kv_init_rootdse,
+    .search = ldb_kv_handle_request,
+    .add = ldb_kv_handle_request,
+    .modify = ldb_kv_handle_request,
+    .del = ldb_kv_handle_request,
+    .rename = ldb_kv_handle_request,
+    .extended = ldb_kv_handle_request,
+    .start_transaction = ldb_kv_start_trans,
+    .end_transaction = ldb_kv_end_trans,
+    .prepare_commit = ldb_kv_prepare_commit,
+    .del_transaction = ldb_kv_del_trans,
+    .read_lock = ldb_kv_lock_read,
+    .read_unlock = ldb_kv_unlock_read,
+};
+
+int ldb_kv_init_store(struct ldb_kv_private *ldb_kv,
+		      const char *name,
+		      struct ldb_context *ldb,
+		      const char *options[],
+		      struct ldb_module **_module)
+{
+	if (getenv("LDB_WARN_UNINDEXED")) {
+		ldb_kv->warn_unindexed = true;
+	}
+
+	if (getenv("LDB_WARN_REINDEX")) {
+		ldb_kv->warn_reindex = true;
+	}
+
+	ldb_kv->sequence_number = 0;
+
+	ldb_kv->pid = getpid();
+
+	ldb_kv->pack_format_override = 0;
+
+	ldb_kv->module = ldb_module_new(ldb, ldb, name, &ldb_kv_ops);
+	if (!ldb_kv->module) {
+		ldb_oom(ldb);
+		talloc_free(ldb_kv);
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+	ldb_module_set_private(ldb_kv->module, ldb_kv);
+	talloc_steal(ldb_kv->module, ldb_kv);
+
+	if (ldb_kv_cache_load(ldb_kv->module) != 0) {
+		ldb_asprintf_errstring(ldb, "Unable to load ltdb cache "
+				       "records for backend '%s'", name);
+		talloc_free(ldb_kv->module);
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	*_module = ldb_kv->module;
+	/*
+	 * Set or override the maximum key length
+	 *
+	 * The ldb_mdb code will have set this to 511, but our tests
+	 * set this even smaller (to make the tests more practical).
+	 *
+	 * This must only be used for the selftest as the length
+	 * becomes encoded in the index keys.
+	 */
+	{
+		const char *len_str =
+			ldb_options_find(ldb, options,
+					 "max_key_len_for_self_test");
+		if (len_str != NULL) {
+			unsigned len = strtoul(len_str, NULL, 0);
+			ldb_kv->max_key_length = len;
+		}
+	}
+
+	/*
+	 * Usually the presence of GUID indexing determines the pack format
+	 * we use but in certain circumstances such as downgrading an
+	 * MDB-backed database, we want to override the target pack format.
+	 *
+	 * We set/get opaques here because in the Samba partitions module,
+	 * 'options' are not passed correctly so sub-databases can't see
+	 * the options they need.
+	 */
+	{
+		const char *pack_format_override =
+			ldb_options_find(ldb, options, "pack_format_override");
+		if (pack_format_override != NULL) {
+			int ret;
+			ldb_kv->pack_format_override =
+				strtoul(pack_format_override, NULL, 0);
+			ret = ldb_set_opaque(ldb,
+					     "pack_format_override",
+			     (void *)(intptr_t)ldb_kv->pack_format_override);
+			if (ret != LDB_SUCCESS) {
+				talloc_free(ldb_kv->module);
+				return ldb_module_operr(ldb_kv->module);
+			}
+		} else {
+			/*
+			 * NULL -> 0 is fine, otherwise we get back
+			 * the number we needed
+			 */
+			ldb_kv->pack_format_override
+				= (intptr_t)ldb_get_opaque(ldb,
+						   "pack_format_override");
+		}
+	}
+
+	/*
+	 * Override full DB scans
+	 *
+	 * A full DB scan is expensive on a large database.  This
+	 * option is for testing to show that the full DB scan is not
+	 * triggered.
+	 */
+	{
+		const char *len_str =
+			ldb_options_find(ldb, options,
+					 "disable_full_db_scan_for_self_test");
+		if (len_str != NULL) {
+			ldb_kv->disable_full_db_scan = true;
+		}
+	}
+
+	/*
+	 * Set the size of the transaction index cache.
+	 * If the ldb option "transaction_index_cache_size" is set use that
+	 * otherwise use DEFAULT_INDEX_CACHE_SIZE
+	 */
+	ldb_kv->index_transaction_cache_size = DEFAULT_INDEX_CACHE_SIZE;
+	{
+		const char *size = ldb_options_find(
+			ldb,
+			options,
+			"transaction_index_cache_size");
+		if (size != NULL) {
+			size_t cache_size = 0;
+			errno = 0;
+
+			cache_size = strtoul( size, NULL, 0);
+			if (cache_size == 0 || errno == ERANGE) {
+				ldb_debug(
+					ldb,
+					LDB_DEBUG_WARNING,
+					"Invalid transaction_index_cache_size "
+					"value [%s], using default(%d)\n",
+					size,
+					DEFAULT_INDEX_CACHE_SIZE);
+			} else {
+				ldb_kv->index_transaction_cache_size =
+					cache_size;
+			}
+		}
+	}
+	/*
+	 * Set batch mode operation.
+	 * This disables the nested sub transactions, and increases the
+	 * chance of index corruption.  If using this mode the transaction
+	 * commit will be aborted if any operation fails.
+	 */
+	{
+		const char *batch_mode = ldb_options_find(
+			ldb, options, "batch_mode");
+		if (batch_mode != NULL) {
+			ldb_kv->batch_mode = true;
+		}
+	}
+
+	return LDB_SUCCESS;
+}
diff --git a/lib/ldb/ldb_key_value/ldb_kv.h b/lib/ldb/ldb_key_value/ldb_kv.h
new file mode 100644
index 0000000..78cbbec
--- /dev/null
+++ b/lib/ldb/ldb_key_value/ldb_kv.h
@@ -0,0 +1,344 @@
+#include "replace.h"
+#include "system/filesys.h"
+#include "system/time.h"
+#include "tdb.h"
+#include "ldb_module.h"
+
+#ifndef __LDB_KV_H__
+#define __LDB_KV_H__
+struct ldb_kv_private;
+typedef int (*ldb_kv_traverse_fn)(struct ldb_kv_private *ldb_kv,
+				  struct ldb_val key,
+				  struct ldb_val data,
+				  void *ctx);
+
+struct kv_db_ops {
+	uint32_t options;
+
+	int (*store)(struct ldb_kv_private *ldb_kv,
+		     struct ldb_val key,
+		     struct ldb_val data,
+		     int flags);
+	int (*delete)(struct ldb_kv_private *ldb_kv, struct ldb_val key);
+	int (*iterate)(struct ldb_kv_private *ldb_kv,
+		       ldb_kv_traverse_fn fn,
+		       void *ctx);
+	int (*update_in_iterate)(struct ldb_kv_private *ldb_kv,
+				 struct ldb_val key,
+				 struct ldb_val key2,
+				 struct ldb_val data,
+				 void *ctx);
+	int (*fetch_and_parse)(struct ldb_kv_private *ldb_kv,
+			       struct ldb_val key,
+			       int (*parser)(struct ldb_val key,
+					     struct ldb_val data,
+					     void *private_data),
+			       void *ctx);
+	int (*iterate_range)(struct ldb_kv_private *ldb_kv,
+			     struct ldb_val start_key,
+			     struct ldb_val end_key,
+			     ldb_kv_traverse_fn fn,
+			     void *ctx);
+	int (*lock_read)(struct ldb_module *);
+	int (*unlock_read)(struct ldb_module *);
+	int (*begin_write)(struct ldb_kv_private *);
+	int (*prepare_write)(struct ldb_kv_private *);
+	int (*abort_write)(struct ldb_kv_private *);
+	int (*finish_write)(struct ldb_kv_private *);
+	int (*error)(struct ldb_kv_private *ldb_kv);
+	const char *(*errorstr)(struct ldb_kv_private *ldb_kv);
+	const char *(*name)(struct ldb_kv_private *ldb_kv);
+	bool (*has_changed)(struct ldb_kv_private *ldb_kv);
+	bool (*transaction_active)(struct ldb_kv_private *ldb_kv);
+	size_t (*get_size)(struct ldb_kv_private *ldb_kv);
+	int (*begin_nested_write)(struct ldb_kv_private *);
+	int (*finish_nested_write)(struct ldb_kv_private *);
+	int (*abort_nested_write)(struct ldb_kv_private *);
+};
+
+/* this private structure is used by the key value backends in the
+   ldb_context */
+struct ldb_kv_private {
+	const struct kv_db_ops *kv_ops;
+	struct ldb_module *module;
+	TDB_CONTEXT *tdb;
+	struct lmdb_private *lmdb_private;
+	unsigned int connect_flags;
+
+	unsigned long long sequence_number;
+	uint32_t pack_format_version;
+	uint32_t target_pack_format_version;
+	uint32_t pack_format_override;
+
+	/* the low level tdb seqnum - used to avoid loading BASEINFO when
+	   possible */
+	int tdb_seqnum;
+
+	struct ldb_kv_cache {
+		struct ldb_message *indexlist;
+		bool one_level_indexes;
+		bool attribute_indexes;
+		const char *GUID_index_attribute;
+		const char *GUID_index_dn_component;
+	} *cache;
+
+
+	bool check_base;
+	bool disallow_dn_filter;
+	/*
+	 * To improve the performance of batch operations we maintain a cache
+	 * of index records, these entries get written to disk in the
+	 * prepare_commit phase.
+	 */
+	struct ldb_kv_idxptr *idxptr;
+	/*
+	 * To ensure that the indexes in idxptr are consistent we cache any
+	 * index updates during an operation, i.e. ldb_kv_add, ldb_kv_delete ...
+	 * Once the changes to the data record have been committed to disk
+	 * the contents of this cache are copied to idxptr
+	 */
+	struct ldb_kv_idxptr *nested_idx_ptr;
+	/*
+	 * If batch mode is set the sub transactions and index caching
+	 * wrapping individual operations is disabled.
+	 * This is to improve the performance of large batch operations
+	 * i.e. domain joins.
+	 */
+	bool batch_mode;
+	/*
+	 * Has an operation failed, if true and we're in batch_mode
+	 * the transaction commit will fail.
+	 */
+	bool operation_failed;
+
+	bool prepared_commit;
+	int read_lock_count;
+
+	bool warn_unindexed;
+	bool warn_reindex;
+
+	bool read_only;
+
+	bool reindex_failed;
+
+	const struct ldb_schema_syntax *GUID_index_syntax;
+
+	/*
+	 * Maximum index key length.  If non zero keys longer than this length
+	 * will be truncated for non unique indexes. Keys for unique indexes
+	 * greater than this length will be rejected.
+	 */
+	unsigned max_key_length;
+
+	/*
+	 * To allow testing that ensures the DB does not fall back
+	 * to a full scan
+	 */
+	bool disable_full_db_scan;
+
+	/*
+	 * The PID that opened this database so we don't work in a
+	 * fork()ed child.
+	 */
+	pid_t pid;
+
+	/*
+	 * The size to be used for the index transaction cache
+	 */
+	size_t index_transaction_cache_size;
+};
+
+struct ldb_kv_context {
+	struct ldb_module *module;
+	struct ldb_request *req;
+
+	/*
+	 * Required as we might not get to the event loop before the
+	 * timeout, so we need some old-style cooperative multitasking
+	 * here.
+	 */
+	struct timeval timeout_timeval;
+
+	/* Used to throttle calls to gettimeofday() */
+	size_t timeout_counter;
+
+	bool request_terminated;
+	struct ldb_kv_req_spy *spy;
+
+	/* search stuff */
+	const struct ldb_parse_tree *tree;
+	struct ldb_dn *base;
+	enum ldb_scope scope;
+	const char * const *attrs;
+	struct tevent_timer *timeout_event;
+
+	/* error handling */
+	int error;
+};
+
+struct ldb_kv_reindex_context {
+	int error;
+	uint32_t count;
+};
+
+struct ldb_kv_repack_context {
+	int error;
+	uint32_t count;
+	bool normal_record_seen;
+	uint32_t old_version;
+};
+
+
+/* special record types */
+#define LDB_KV_INDEX      "@INDEX"
+#define LDB_KV_INDEXLIST  "@INDEXLIST"
+#define LDB_KV_IDX        "@IDX"
+#define LDB_KV_IDXVERSION "@IDXVERSION"
+#define LDB_KV_IDXATTR    "@IDXATTR"
+#define LDB_KV_IDXONE     "@IDXONE"
+#define LDB_KV_IDXDN     "@IDXDN"
+#define LDB_KV_IDXGUID    "@IDXGUID"
+#define LDB_KV_IDX_DN_GUID "@IDX_DN_GUID"
+
+/*
+ * This will be used to indicate when a new, yet to be developed
+ * sub-database version of the indices are in use, to ensure we do
+ * not load future databases unintentionally.
+ */
+
+#define LDB_KV_IDX_LMDB_SUBDB "@IDX_LMDB_SUBDB"
+
+#define LDB_KV_BASEINFO   "@BASEINFO"
+#define LDB_KV_OPTIONS    "@OPTIONS"
+#define LDB_KV_ATTRIBUTES "@ATTRIBUTES"
+
+/* special attribute types */
+#define LDB_KV_SEQUENCE_NUMBER "sequenceNumber"
+#define LDB_KV_CHECK_BASE "checkBaseOnSearch"
+#define LDB_KV_DISALLOW_DN_FILTER "disallowDNFilter"
+#define LDB_KV_MOD_TIMESTAMP "whenChanged"
+#define LDB_KV_OBJECTCLASS "objectClass"
+
+/* DB keys */
+#define LDB_KV_GUID_KEY_PREFIX "GUID="
+#define LDB_KV_GUID_SIZE 16
+#define LDB_KV_GUID_KEY_SIZE (LDB_KV_GUID_SIZE + sizeof(LDB_KV_GUID_KEY_PREFIX) - 1)
+
+/* LDB KV options */
+/*
+ * This allows pointers to be referenced after the callback to any variant of
+ * iterate or fetch_and_parse -- as long as an overall read lock is held.
+ */
+#define LDB_KV_OPTION_STABLE_READ_LOCK 0x00000001
+
+/*
+ * The following definitions come from lib/ldb/ldb_key_value/ldb_kv_cache.c
+ */
+
+int ldb_kv_cache_reload(struct ldb_module *module);
+int ldb_kv_cache_load(struct ldb_module *module);
+int ldb_kv_increase_sequence_number(struct ldb_module *module);
+int ldb_kv_check_at_attributes_values(const struct ldb_val *value);
+
+/*
+ * The following definitions come from lib/ldb/ldb_key_value/ldb_kv_index.c
+ */
+
+/*
+ * The default size of the in memory TDB used to cache index records
+ * The value chosen gives a prime modulo for the hash table and keeps the
+ * tdb memory overhead under 4 kB
+ */
+#define DEFAULT_INDEX_CACHE_SIZE 491
+
+struct ldb_parse_tree;
+
+int ldb_kv_search_indexed(struct ldb_kv_context *ctx, uint32_t *);
+int ldb_kv_index_add_new(struct ldb_module *module,
+			 struct ldb_kv_private *ldb_kv,
+			 const struct ldb_message *msg);
+int ldb_kv_index_delete(struct ldb_module *module,
+			const struct ldb_message *msg);
+int ldb_kv_index_del_element(struct ldb_module *module,
+			     struct ldb_kv_private *ldb_kv,
+			     const struct ldb_message *msg,
+			     struct ldb_message_element *el);
+int ldb_kv_index_add_element(struct ldb_module *module,
+			     struct ldb_kv_private *ldb_kv,
+			     const struct ldb_message *msg,
+			     struct ldb_message_element *el);
+int ldb_kv_index_del_value(struct ldb_module *module,
+			   struct ldb_kv_private *ldb_kv,
+			   const struct ldb_message *msg,
+			   struct ldb_message_element *el,
+			   unsigned int v_idx);
+int ldb_kv_reindex(struct ldb_module *module);
+int ldb_kv_repack(struct ldb_module *module);
+int ldb_kv_index_transaction_start(
+	struct ldb_module *module,
+	size_t cache_size);
+int ldb_kv_index_transaction_commit(struct ldb_module *module);
+int ldb_kv_index_transaction_cancel(struct ldb_module *module);
+int ldb_kv_key_dn_from_idx(struct ldb_module *module,
+			   struct ldb_kv_private *ldb_kv,
+			   TALLOC_CTX *mem_ctx,
+			   struct ldb_dn *dn,
+			  struct ldb_val *key);
+
+/*
+ * The following definitions come from lib/ldb/ldb_key_value/ldb_kv_search.c
+ */
+int ldb_kv_search_dn1(struct ldb_module *module,
+		      struct ldb_dn *dn,
+		      struct ldb_message *msg,
+		      unsigned int unpack_flags);
+int ldb_kv_search_base(struct ldb_module *module,
+		       TALLOC_CTX *mem_ctx,
+		       struct ldb_dn *dn,
+		       struct ldb_dn **ret_dn);
+int ldb_kv_search_key(struct ldb_module *module,
+		      struct ldb_kv_private *ldb_kv,
+		      const struct ldb_val ldb_key,
+		      struct ldb_message *msg,
+		      unsigned int unpack_flags);
+int ldb_kv_filter_attrs_in_place(struct ldb_message *msg,
+				 const char *const *attrs);
+int ldb_kv_search(struct ldb_kv_context *ctx);
+
+/*
+ * The following definitions come from lib/ldb/ldb_key_value/ldb_kv.c  */
+/*
+ * Determine if this key could hold a normal record.  We allow the new
+ * GUID index, the old DN index and a possible future ID= but not
+ * DN=@.
+ */
+bool ldb_kv_key_is_normal_record(struct ldb_val key);
+struct ldb_val ldb_kv_key_dn(TALLOC_CTX *mem_ctx,
+			     struct ldb_dn *dn);
+struct ldb_val ldb_kv_key_msg(struct ldb_module *module,
+			     TALLOC_CTX *mem_ctx,
+			      const struct ldb_message *msg);
+int ldb_kv_guid_to_key(const struct ldb_val *GUID_val,
+		       struct ldb_val *key);
+int ldb_kv_idx_to_key(struct ldb_module *module,
+		      struct ldb_kv_private *ldb_kv,
+		      TALLOC_CTX *mem_ctx,
+		      const struct ldb_val *idx_val,
+		      struct ldb_val *key);
+int ldb_kv_store(struct ldb_module *module,
+		 const struct ldb_message *msg,
+		 int flgs);
+int ldb_kv_modify_internal(struct ldb_module *module,
+			   const struct ldb_message *msg,
+			   struct ldb_request *req);
+int ldb_kv_delete_noindex(struct ldb_module *module,
+			  const struct ldb_message *msg);
+int ldb_kv_init_store(struct ldb_kv_private *ldb_kv,
+		      const char *name,
+		      struct ldb_context *ldb,
+		      const char *options[],
+		      struct ldb_module **_module);
+int ldb_kv_index_sub_transaction_start(struct ldb_kv_private *ldb_kv);
+int ldb_kv_index_sub_transaction_cancel(struct ldb_kv_private *ldb_kv);
+int ldb_kv_index_sub_transaction_commit(struct ldb_kv_private *ldb_kv);
+#endif /* __LDB_KV_H__ */
diff --git a/lib/ldb/ldb_key_value/ldb_kv_cache.c b/lib/ldb/ldb_key_value/ldb_kv_cache.c
new file mode 100644
index 0000000..b1c136d
--- /dev/null
+++ b/lib/ldb/ldb_key_value/ldb_kv_cache.c
@@ -0,0 +1,697 @@
+/*
+   ldb database library
+
+   Copyright (C) Andrew Tridgell  2004
+
+     ** NOTE! The following LGPL license applies to the ldb
+     ** library. This does NOT imply that all of Samba is released
+     ** under the LGPL
+
+   This library is free software; you can redistribute it and/or
+   modify it under the terms of the GNU Lesser General Public
+   License as published by the Free Software Foundation; either
+   version 3 of the License, or (at your option) any later version.
+
+   This library is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+   Lesser General Public License for more details.
+
+   You should have received a copy of the GNU Lesser General Public
+   License along with this library; if not, see <http://www.gnu.org/licenses/>.
+*/
+
+/*
+ *  Name: ldb
+ *
+ *  Component: ldb key value cache functions
+ *
+ *  Description: cache special records in a ldb/tdb
+ *
+ *  Author: Andrew Tridgell
+ */
+
+#include "ldb_kv.h"
+#include "ldb_private.h"
+
+#define LDB_KV_FLAG_CASE_INSENSITIVE (1<<0)
+#define LDB_KV_FLAG_INTEGER          (1<<1)
+#define LDB_KV_FLAG_UNIQUE_INDEX     (1<<2)
+#define LDB_KV_FLAG_ORDERED_INTEGER  (1<<3)
+
+/* valid attribute flags */
+static const struct {
+	const char *name;
+	int value;
+} ldb_kv_valid_attr_flags[] = {
+	{ "CASE_INSENSITIVE", LDB_KV_FLAG_CASE_INSENSITIVE },
+	{ "INTEGER", LDB_KV_FLAG_INTEGER },
+	{ "ORDERED_INTEGER", LDB_KV_FLAG_ORDERED_INTEGER },
+	{ "HIDDEN", 0 },
+	{ "UNIQUE_INDEX",  LDB_KV_FLAG_UNIQUE_INDEX},
+	{ "NONE", 0 },
+	{ NULL, 0 }
+};
+
+/*
+  de-register any special handlers for @ATTRIBUTES
+*/
+static void ldb_kv_attributes_unload(struct ldb_module *module)
+{
+	struct ldb_context *ldb = ldb_module_get_ctx(module);
+
+	ldb_schema_attribute_remove_flagged(ldb, LDB_ATTR_FLAG_FROM_DB);
+
+}
+
+/*
+  add up the attrib flags for a @ATTRIBUTES element
+*/
+static int ldb_kv_attributes_flags(struct ldb_message_element *el, unsigned *v)
+{
+	unsigned int i;
+	unsigned value = 0;
+	for (i=0;i<el->num_values;i++) {
+		unsigned int j;
+		for (j = 0; ldb_kv_valid_attr_flags[j].name; j++) {
+			if (strcmp(ldb_kv_valid_attr_flags[j].name,
+				   (char *)el->values[i].data) == 0) {
+				value |= ldb_kv_valid_attr_flags[j].value;
+				break;
+			}
+		}
+		if (ldb_kv_valid_attr_flags[j].name == NULL) {
+			return -1;
+		}
+	}
+	*v = value;
+	return 0;
+}
+
+static int ldb_schema_attribute_compare(const void *p1, const void *p2)
+{
+	const struct ldb_schema_attribute *sa1 = (const struct ldb_schema_attribute *)p1;
+	const struct ldb_schema_attribute *sa2 = (const struct ldb_schema_attribute *)p2;
+	return ldb_attr_cmp(sa1->name, sa2->name);
+}
+
+/*
+  register any special handlers from @ATTRIBUTES
+*/
+static int ldb_kv_attributes_load(struct ldb_module *module)
+{
+	struct ldb_schema_attribute *attrs;
+	struct ldb_context *ldb;
+	struct ldb_message *attrs_msg = NULL;
+	struct ldb_dn *dn;
+	unsigned int i;
+	unsigned int num_loaded_attrs = 0;
+	int r;
+
+	ldb = ldb_module_get_ctx(module);
+
+	if (ldb->schema.attribute_handler_override) {
+		/* we skip loading the @ATTRIBUTES record when a module is supplying
+		   its own attribute handling */
+		return 0;
+	}
+
+	attrs_msg = ldb_msg_new(module);
+	if (attrs_msg == NULL) {
+		goto failed;
+	}
+
+	dn = ldb_dn_new(module, ldb, LDB_KV_ATTRIBUTES);
+	if (dn == NULL) goto failed;
+
+	r = ldb_kv_search_dn1(module,
+			      dn,
+			      attrs_msg,
+			      LDB_UNPACK_DATA_FLAG_NO_VALUES_ALLOC |
+				  LDB_UNPACK_DATA_FLAG_NO_DN);
+	talloc_free(dn);
+	if (r != LDB_SUCCESS && r != LDB_ERR_NO_SUCH_OBJECT) {
+		goto failed;
+	}
+	if (r == LDB_ERR_NO_SUCH_OBJECT || attrs_msg->num_elements == 0) {
+		TALLOC_FREE(attrs_msg);
+		return 0;
+	}
+
+	attrs = talloc_array(attrs_msg,
+			     struct ldb_schema_attribute,
+			     attrs_msg->num_elements
+			     + ldb->schema.num_attributes);
+	if (attrs == NULL) {
+		goto failed;
+	}
+
+	memcpy(attrs,
+	       ldb->schema.attributes,
+	       sizeof(ldb->schema.attributes[0]) * ldb->schema.num_attributes);
+
+	/* mapping these flags onto ldap 'syntaxes' isn't strictly correct,
+	   but its close enough for now */
+	for (i=0;i<attrs_msg->num_elements;i++) {
+		unsigned flags = 0, attr_flags = 0;
+		const char *syntax;
+		const struct ldb_schema_syntax *s;
+		const struct ldb_schema_attribute *a =
+			ldb_schema_attribute_by_name(ldb,
+						     attrs_msg->elements[i].name);
+		if (a != NULL && a->flags & LDB_ATTR_FLAG_FIXED) {
+			/* Must already be set in the array, and kept */
+			continue;
+		}
+
+		if (ldb_kv_attributes_flags(&attrs_msg->elements[i], &flags) !=
+		    0) {
+			ldb_debug(ldb, LDB_DEBUG_ERROR,
+				  "Invalid @ATTRIBUTES element for '%s'",
+				  attrs_msg->elements[i].name);
+			goto failed;
+		}
+
+		if (flags & LDB_KV_FLAG_UNIQUE_INDEX) {
+			attr_flags = LDB_ATTR_FLAG_UNIQUE_INDEX;
+		}
+		flags &= ~LDB_KV_FLAG_UNIQUE_INDEX;
+
+		/* These are not currently flags, each is exclusive */
+		if (flags == LDB_KV_FLAG_CASE_INSENSITIVE) {
+			syntax = LDB_SYNTAX_DIRECTORY_STRING;
+		} else if (flags == LDB_KV_FLAG_INTEGER) {
+			syntax = LDB_SYNTAX_INTEGER;
+		} else if (flags == LDB_KV_FLAG_ORDERED_INTEGER) {
+			syntax = LDB_SYNTAX_ORDERED_INTEGER;
+		} else if (flags == 0) {
+			syntax = LDB_SYNTAX_OCTET_STRING;
+		} else {
+			ldb_debug(ldb, LDB_DEBUG_ERROR,
+				  "Invalid flag combination 0x%x for '%s' "
+				  "in @ATTRIBUTES",
+				  flags, attrs_msg->elements[i].name);
+			goto failed;
+		}
+
+		s = ldb_standard_syntax_by_name(ldb, syntax);
+		if (s == NULL) {
+			ldb_debug(ldb, LDB_DEBUG_ERROR,
+				  "Invalid attribute syntax '%s' for '%s' "
+				  "in @ATTRIBUTES",
+				  syntax, attrs_msg->elements[i].name);
+			goto failed;
+		}
+
+		attr_flags |= LDB_ATTR_FLAG_ALLOCATED | LDB_ATTR_FLAG_FROM_DB;
+
+		r = ldb_schema_attribute_fill_with_syntax(ldb,
+							  attrs,
+							  attrs_msg->elements[i].name,
+							  attr_flags, s,
+							  &attrs[num_loaded_attrs + ldb->schema.num_attributes]);
+		if (r != 0) {
+			goto failed;
+		}
+		num_loaded_attrs++;
+	}
+
+	attrs = talloc_realloc(attrs_msg,
+			       attrs, struct ldb_schema_attribute,
+			       num_loaded_attrs + ldb->schema.num_attributes);
+	if (attrs == NULL) {
+		goto failed;
+	}
+	TYPESAFE_QSORT(attrs, num_loaded_attrs + ldb->schema.num_attributes,
+		       ldb_schema_attribute_compare);
+	talloc_unlink(ldb, ldb->schema.attributes);
+	ldb->schema.attributes = talloc_steal(ldb, attrs);
+	ldb->schema.num_attributes = num_loaded_attrs + ldb->schema.num_attributes;
+	TALLOC_FREE(attrs_msg);
+
+	return 0;
+failed:
+	TALLOC_FREE(attrs_msg);
+	return -1;
+}
+
+/*
+  register any index records we find for the DB
+*/
+static int ldb_kv_index_load(struct ldb_module *module,
+			     struct ldb_kv_private *ldb_kv)
+{
+	struct ldb_context *ldb = ldb_module_get_ctx(module);
+	struct ldb_dn *indexlist_dn;
+	int r, lmdb_subdb_version;
+
+	if (ldb->schema.index_handler_override) {
+		/*
+		 * we skip loading the @INDEXLIST record when a module is
+		 * supplying its own attribute handling
+		 */
+		ldb_kv->cache->attribute_indexes = true;
+		ldb_kv->cache->one_level_indexes =
+		    ldb->schema.one_level_indexes;
+		ldb_kv->cache->GUID_index_attribute =
+		    ldb->schema.GUID_index_attribute;
+		ldb_kv->cache->GUID_index_dn_component =
+		    ldb->schema.GUID_index_dn_component;
+		return 0;
+	}
+
+	talloc_free(ldb_kv->cache->indexlist);
+
+	ldb_kv->cache->indexlist = ldb_msg_new(ldb_kv->cache);
+	if (ldb_kv->cache->indexlist == NULL) {
+		return -1;
+	}
+	ldb_kv->cache->one_level_indexes = false;
+	ldb_kv->cache->attribute_indexes = false;
+
+	indexlist_dn = ldb_dn_new(ldb_kv, ldb, LDB_KV_INDEXLIST);
+	if (indexlist_dn == NULL) {
+		return -1;
+	}
+
+	r = ldb_kv_search_dn1(module,
+			      indexlist_dn,
+			      ldb_kv->cache->indexlist,
+			      LDB_UNPACK_DATA_FLAG_NO_VALUES_ALLOC |
+				  LDB_UNPACK_DATA_FLAG_NO_DN);
+	TALLOC_FREE(indexlist_dn);
+
+	if (r != LDB_SUCCESS && r != LDB_ERR_NO_SUCH_OBJECT) {
+		return -1;
+	}
+
+	if (ldb_msg_find_element(ldb_kv->cache->indexlist, LDB_KV_IDXONE) !=
+	    NULL) {
+		ldb_kv->cache->one_level_indexes = true;
+	}
+	if (ldb_msg_find_element(ldb_kv->cache->indexlist, LDB_KV_IDXATTR) !=
+	    NULL) {
+		ldb_kv->cache->attribute_indexes = true;
+	}
+	ldb_kv->cache->GUID_index_attribute = ldb_msg_find_attr_as_string(
+	    ldb_kv->cache->indexlist, LDB_KV_IDXGUID, NULL);
+	ldb_kv->cache->GUID_index_dn_component = ldb_msg_find_attr_as_string(
+	    ldb_kv->cache->indexlist, LDB_KV_IDX_DN_GUID, NULL);
+
+	lmdb_subdb_version = ldb_msg_find_attr_as_int(
+	    ldb_kv->cache->indexlist, LDB_KV_IDX_LMDB_SUBDB, 0);
+
+	if (lmdb_subdb_version != 0) {
+		ldb_set_errstring(ldb,
+				  "FATAL: This ldb_mdb database has "
+				  "been written in a new version of LDB "
+				  "using a sub-database index that "
+				  "is not understood by ldb "
+				  LDB_VERSION);
+		return -1;
+	}
+
+	return 0;
+}
+
+/*
+  initialise the baseinfo record
+*/
+static int ldb_kv_baseinfo_init(struct ldb_module *module)
+{
+	struct ldb_context *ldb;
+	void *data = ldb_module_get_private(module);
+	struct ldb_kv_private *ldb_kv =
+	    talloc_get_type(data, struct ldb_kv_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
+	   out. */
+	const char *initial_sequence_number = "1";
+
+	ldb = ldb_module_get_ctx(module);
+
+	ldb_kv->sequence_number = atof(initial_sequence_number);
+
+	msg = ldb_msg_new(ldb_kv);
+	if (msg == NULL) {
+		goto failed;
+	}
+
+	msg->num_elements = 1;
+	msg->elements = &el;
+	msg->dn = ldb_dn_new(msg, ldb, LDB_KV_BASEINFO);
+	if (!msg->dn) {
+		goto failed;
+	}
+	el.name = talloc_strdup(msg, LDB_KV_SEQUENCE_NUMBER);
+	if (!el.name) {
+		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) {
+		goto failed;
+	}
+	val.length = 1;
+
+	ret = ldb_kv_store(module, msg, TDB_INSERT);
+
+	talloc_free(msg);
+
+	return ret;
+
+failed:
+	talloc_free(msg);
+	errno = ENOMEM;
+	return LDB_ERR_OPERATIONS_ERROR;
+}
+
+/*
+  free any cache records
+ */
+static void ldb_kv_cache_free(struct ldb_module *module)
+{
+	void *data = ldb_module_get_private(module);
+	struct ldb_kv_private *ldb_kv =
+	    talloc_get_type(data, struct ldb_kv_private);
+
+	ldb_kv->sequence_number = 0;
+	talloc_free(ldb_kv->cache);
+	ldb_kv->cache = NULL;
+}
+
+/*
+  force a cache reload
+*/
+int ldb_kv_cache_reload(struct ldb_module *module)
+{
+	ldb_kv_attributes_unload(module);
+	ldb_kv_cache_free(module);
+	return ldb_kv_cache_load(module);
+}
+static int get_pack_format_version(struct ldb_val key,
+				   struct ldb_val data,
+				   void *private_data)
+{
+	uint32_t *v = (uint32_t *) private_data;
+	return ldb_unpack_get_format(&data, v);
+}
+
+/*
+  load the cache records
+*/
+int ldb_kv_cache_load(struct ldb_module *module)
+{
+	struct ldb_context *ldb;
+	void *data = ldb_module_get_private(module);
+	struct ldb_kv_private *ldb_kv =
+	    talloc_get_type(data, struct ldb_kv_private);
+	struct ldb_dn *baseinfo_dn = NULL, *options_dn = NULL;
+	uint64_t seq;
+	struct ldb_message *baseinfo = NULL, *options = NULL;
+	const struct ldb_schema_attribute *a;
+	bool have_write_txn = false;
+	int r;
+	struct ldb_val key;
+
+	ldb = ldb_module_get_ctx(module);
+
+	/* a very fast check to avoid extra database reads */
+	if (ldb_kv->cache != NULL && !ldb_kv->kv_ops->has_changed(ldb_kv)) {
+		return 0;
+	}
+
+	if (ldb_kv->cache == NULL) {
+		ldb_kv->cache = talloc_zero(ldb_kv, struct ldb_kv_cache);
+		if (ldb_kv->cache == NULL)
+			goto failed;
+	}
+
+	baseinfo = ldb_msg_new(ldb_kv->cache);
+	if (baseinfo == NULL) goto failed;
+
+	baseinfo_dn = ldb_dn_new(baseinfo, ldb, LDB_KV_BASEINFO);
+	if (baseinfo_dn == NULL) goto failed;
+
+	r = ldb_kv->kv_ops->lock_read(module);
+	if (r != LDB_SUCCESS) {
+		goto failed;
+	}
+
+	key = ldb_kv_key_dn(baseinfo, baseinfo_dn);
+	if (!key.data) {
+		goto failed_and_unlock;
+	}
+
+	/* Read packing format from first 4 bytes of @BASEINFO record */
+	r = ldb_kv->kv_ops->fetch_and_parse(ldb_kv, key,
+					    get_pack_format_version,
+					    &ldb_kv->pack_format_version);
+
+	/* possibly initialise the baseinfo */
+	if (r == LDB_ERR_NO_SUCH_OBJECT) {
+
+		/* Give up the read lock, try again with a write lock */
+		r = ldb_kv->kv_ops->unlock_read(module);
+		if (r != LDB_SUCCESS) {
+			goto failed;
+		}
+
+		if (ldb_kv->kv_ops->begin_write(ldb_kv) != 0) {
+			goto failed;
+		}
+
+		have_write_txn = true;
+
+		/*
+		 * We need to write but haven't figured out packing format yet.
+		 * Just go with version 1 and we'll repack if we got it wrong.
+		 */
+		ldb_kv->pack_format_version = LDB_PACKING_FORMAT;
+		ldb_kv->target_pack_format_version = LDB_PACKING_FORMAT;
+
+		/* error handling for ltdb_baseinfo_init() is by
+		   looking for the record again. */
+		ldb_kv_baseinfo_init(module);
+
+	} else if (r != LDB_SUCCESS) {
+		goto failed_and_unlock;
+	}
+
+	/* OK now we definitely have a @BASEINFO record so fetch it */
+	r = ldb_kv_search_dn1(module, baseinfo_dn, baseinfo, 0);
+	if (r != LDB_SUCCESS) {
+		goto failed_and_unlock;
+	}
+
+	/* Ignore the result, and update the sequence number */
+	ldb_kv->kv_ops->has_changed(ldb_kv);
+
+	/* if the current internal sequence number is the same as the one
+	   in the database then assume the rest of the cache is OK */
+	seq = ldb_msg_find_attr_as_uint64(baseinfo, LDB_KV_SEQUENCE_NUMBER, 0);
+	if (seq == ldb_kv->sequence_number) {
+		goto done;
+	}
+	ldb_kv->sequence_number = seq;
+
+	/* Read an interpret database options */
+
+	options = ldb_msg_new(ldb_kv->cache);
+	if (options == NULL) goto failed_and_unlock;
+
+	options_dn = ldb_dn_new(options, ldb, LDB_KV_OPTIONS);
+	if (options_dn == NULL) goto failed_and_unlock;
+
+	r = ldb_kv_search_dn1(module, options_dn, options, 0);
+	talloc_free(options_dn);
+	if (r != LDB_SUCCESS && r != LDB_ERR_NO_SUCH_OBJECT) {
+		goto failed_and_unlock;
+	}
+
+	/* set flags if they do exist */
+	if (r == LDB_SUCCESS) {
+		ldb_kv->check_base =
+		    ldb_msg_find_attr_as_bool(options, LDB_KV_CHECK_BASE, false);
+		ldb_kv->disallow_dn_filter = ldb_msg_find_attr_as_bool(
+		    options, LDB_KV_DISALLOW_DN_FILTER, false);
+	} else {
+		ldb_kv->check_base = false;
+		ldb_kv->disallow_dn_filter = false;
+	}
+
+	/*
+	 * ltdb_attributes_unload() calls internally talloc_free() on
+	 * any non-fixed elements in ldb->schema.attributes.
+	 *
+	 * NOTE WELL: This is per-ldb, not per module, so overwrites
+	 * the handlers across all databases when used under Samba's
+	 * partition module.
+	 */
+	ldb_kv_attributes_unload(module);
+
+	if (ldb_kv_index_load(module, ldb_kv) == -1) {
+		goto failed_and_unlock;
+	}
+
+	/*
+	 * NOTE WELL: This is per-ldb, not per module, so overwrites
+	 * the handlers across all databases when used under Samba's
+	 * partition module.
+	 */
+	if (ldb_kv_attributes_load(module) == -1) {
+		goto failed_and_unlock;
+	}
+
+	/*
+	 * Initialise packing version and GUID index syntax, and force the
+	 * two to travel together, ie a GUID indexed database must use V2
+	 * packing format and a DN indexed database must use V1.
+	 */
+	ldb_kv->GUID_index_syntax = NULL;
+	if (ldb_kv->cache->GUID_index_attribute != NULL) {
+		ldb_kv->target_pack_format_version = LDB_PACKING_FORMAT_V2;
+
+		/*
+		 * Now the attributes are loaded, set the guid_index_syntax.
+		 * This can't fail, it will return a default at worst
+		 */
+		a = ldb_schema_attribute_by_name(
+		    ldb, ldb_kv->cache->GUID_index_attribute);
+		ldb_kv->GUID_index_syntax = a->syntax;
+	} else {
+		ldb_kv->target_pack_format_version = LDB_PACKING_FORMAT;
+	}
+
+done:
+	if (have_write_txn) {
+		if (ldb_kv->kv_ops->finish_write(ldb_kv) != 0) {
+			goto failed;
+		}
+	} else {
+		ldb_kv->kv_ops->unlock_read(module);
+	}
+
+	talloc_free(options);
+	talloc_free(baseinfo);
+	return 0;
+
+failed_and_unlock:
+	if (have_write_txn) {
+		ldb_kv->kv_ops->abort_write(ldb_kv);
+	} else {
+		ldb_kv->kv_ops->unlock_read(module);
+	}
+
+failed:
+	talloc_free(options);
+	talloc_free(baseinfo);
+	return -1;
+}
+
+
+/*
+  increase the sequence number to indicate a database change
+*/
+int ldb_kv_increase_sequence_number(struct ldb_module *module)
+{
+	struct ldb_context *ldb;
+	void *data = ldb_module_get_private(module);
+	struct ldb_kv_private *ldb_kv =
+	    talloc_get_type(data, struct ldb_kv_private);
+	struct ldb_message *msg;
+	struct ldb_message_element el[2];
+	struct ldb_val val;
+	struct ldb_val val_time;
+	time_t t = time(NULL);
+	char *s = NULL;
+	int ret;
+
+	ldb = ldb_module_get_ctx(module);
+
+	msg = ldb_msg_new(ldb_kv);
+	if (msg == NULL) {
+		errno = ENOMEM;
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	s = talloc_asprintf(msg, "%llu", ldb_kv->sequence_number + 1);
+	if (!s) {
+		talloc_free(msg);
+		errno = ENOMEM;
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	msg->num_elements = ARRAY_SIZE(el);
+	msg->elements = el;
+	msg->dn = ldb_dn_new(msg, ldb, LDB_KV_BASEINFO);
+	if (msg->dn == NULL) {
+		talloc_free(msg);
+		errno = ENOMEM;
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+	el[0].name = talloc_strdup(msg, LDB_KV_SEQUENCE_NUMBER);
+	if (el[0].name == NULL) {
+		talloc_free(msg);
+		errno = ENOMEM;
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+	el[0].values = &val;
+	el[0].num_values = 1;
+	el[0].flags = LDB_FLAG_MOD_REPLACE;
+	val.data = (uint8_t *)s;
+	val.length = strlen(s);
+
+	el[1].name = talloc_strdup(msg, LDB_KV_MOD_TIMESTAMP);
+	if (el[1].name == NULL) {
+		talloc_free(msg);
+		errno = ENOMEM;
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+	el[1].values = &val_time;
+	el[1].num_values = 1;
+	el[1].flags = LDB_FLAG_MOD_REPLACE;
+
+	s = ldb_timestring(msg, t);
+	if (s == NULL) {
+		talloc_free(msg);
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	val_time.data = (uint8_t *)s;
+	val_time.length = strlen(s);
+
+	ret = ldb_kv_modify_internal(module, msg, NULL);
+
+	talloc_free(msg);
+
+	if (ret == LDB_SUCCESS) {
+		ldb_kv->sequence_number += 1;
+	}
+
+	/* updating the tdb_seqnum here avoids us reloading the cache
+	   records due to our own modification */
+	ldb_kv->kv_ops->has_changed(ldb_kv);
+
+	return ret;
+}
+
+int ldb_kv_check_at_attributes_values(const struct ldb_val *value)
+{
+	unsigned int i;
+
+	for (i = 0; ldb_kv_valid_attr_flags[i].name != NULL; i++) {
+		if ((strcmp(ldb_kv_valid_attr_flags[i].name,
+			    (char *)value->data) == 0)) {
+			return 0;
+		}
+	}
+
+	return -1;
+}
diff --git a/lib/ldb/ldb_key_value/ldb_kv_index.c b/lib/ldb/ldb_key_value/ldb_kv_index.c
new file mode 100644
index 0000000..3f1a847
--- /dev/null
+++ b/lib/ldb/ldb_key_value/ldb_kv_index.c
@@ -0,0 +1,4018 @@
+/*
+   ldb database library
+
+   Copyright (C) Andrew Tridgell  2004-2009
+
+     ** NOTE! The following LGPL license applies to the ldb
+     ** library. This does NOT imply that all of Samba is released
+     ** under the LGPL
+
+   This library is free software; you can redistribute it and/or
+   modify it under the terms of the GNU Lesser General Public
+   License as published by the Free Software Foundation; either
+   version 3 of the License, or (at your option) any later version.
+
+   This library is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+   Lesser General Public License for more details.
+
+   You should have received a copy of the GNU Lesser General Public
+   License along with this library; if not, see <http://www.gnu.org/licenses/>.
+*/
+
+/*
+ *  Name: ldb
+ *
+ *  Component: ldb key value backend - indexing
+ *
+ *  Description: indexing routines for ldb key value backend
+ *
+ *  Author: Andrew Tridgell
+ */
+
+/*
+
+LDB Index design and choice of key:
+=======================================
+
+LDB has index records held as LDB objects with a special record like:
+
+dn: @INDEX:attr:value
+
+value may be base64 encoded, if it is deemed not printable:
+
+dn: @INDEX:attr::base64-value
+
+In each record, there is two possible formats:
+
+The original format is:
+-----------------------
+
+dn: @INDEX:NAME:DNSUPDATEPROXY
+@IDXVERSION: 2
+@IDX: CN=DnsUpdateProxy,CN=Users,DC=addom,DC=samba,DC=example,DC=com
+
+In this format, @IDX is multi-valued, one entry for each match
+
+The corresponding entry is stored in a TDB record with key:
+
+DN=CN=DNSUPDATEPROXY,CN=USERS,DC=ADDOM,DC=SAMBA,DC=EXAMPLE,DC=COM
+
+(This allows a scope BASE search to directly find the record via
+a simple casefold of the DN).
+
+The original mixed-case DN is stored in the entry itself.
+
+
+The new 'GUID index' format is:
+-------------------------------
+
+dn: @INDEX:NAME:DNSUPDATEPROXY
+@IDXVERSION: 3
+@IDX: <binary GUID>[<binary GUID>[...]]
+
+The binary guid is 16 bytes, as bytes and not expanded as hexadecimal
+or pretty-printed.  The GUID is chosen from the message to be stored
+by the @IDXGUID attribute on @INDEXLIST.
+
+If there are multiple values the @IDX value simply becomes longer,
+in multiples of 16.
+
+The corresponding entry is stored in a TDB record with key:
+
+GUID=<binary GUID>
+
+This allows a very quick translation between the fixed-length index
+values and the TDB key, while separating entries from other data
+in the TDB, should they be unlucky enough to start with the bytes of
+the 'DN=' prefix.
+
+Additionally, this allows a scope BASE search to directly find the
+record via a simple match on a GUID= extended DN, controlled via
+@IDX_DN_GUID on @INDEXLIST
+
+Exception for special @ DNs:
+
+@BASEINFO, @INDEXLIST and all other special DNs are stored as per the
+original format, as they are never referenced in an index and are used
+to bootstrap the database.
+
+
+Control points for choice of index mode
+---------------------------------------
+
+The choice of index and TDB key mode is made based (for example, from
+Samba) on entries in the @INDEXLIST DN:
+
+dn: @INDEXLIST
+@IDXGUID: objectGUID
+@IDX_DN_GUID: GUID
+
+By default, the original DN format is used.
+
+
+Control points for choosing indexed attributes
+----------------------------------------------
+
+@IDXATTR controls if an attribute is indexed
+
+dn: @INDEXLIST
+@IDXATTR: samAccountName
+@IDXATTR: nETBIOSName
+
+
+C Override functions
+--------------------
+
+void ldb_schema_set_override_GUID_index(struct ldb_context *ldb,
+                                        const char *GUID_index_attribute,
+                                        const char *GUID_index_dn_component)
+
+This is used, particularly in combination with the below, instead of
+the @IDXGUID and @IDX_DN_GUID values in @INDEXLIST.
+
+void ldb_schema_set_override_indexlist(struct ldb_context *ldb,
+                                       bool one_level_indexes);
+void ldb_schema_attribute_set_override_handler(struct ldb_context *ldb,
+                                               ldb_attribute_handler_override_fn_t override,
+                                               void *private_data);
+
+When the above two functions are called in combination, the @INDEXLIST
+values are not read from the DB, so
+ldb_schema_set_override_GUID_index() must be called.
+
+*/
+
+#include "ldb_kv.h"
+#include "../ldb_tdb/ldb_tdb.h"
+#include "ldb_private.h"
+#include "lib/util/binsearch.h"
+#include "lib/util/attr.h"
+
+struct dn_list {
+	unsigned int count;
+	struct ldb_val *dn;
+	/*
+	 * Do not optimise the intersection of this list,
+	 * we must never return an entry not in this
+	 * list.  This allows the index for
+	 * SCOPE_ONELEVEL to be trusted.
+	 */
+	bool strict;
+};
+
+struct ldb_kv_idxptr {
+	/*
+	 * In memory tdb to cache the index updates performed during a
+	 * transaction.  This improves the performance of operations like
+	 * re-index and join
+	 */
+	struct tdb_context *itdb;
+	int error;
+};
+
+enum key_truncation {
+	KEY_NOT_TRUNCATED,
+	KEY_TRUNCATED,
+};
+
+static int ldb_kv_write_index_dn_guid(struct ldb_module *module,
+				      const struct ldb_message *msg,
+				      int add);
+static int ldb_kv_index_dn_base_dn(struct ldb_module *module,
+				   struct ldb_kv_private *ldb_kv,
+				   struct ldb_dn *base_dn,
+				   struct dn_list *dn_list,
+				   enum key_truncation *truncation);
+
+static void ldb_kv_dn_list_sort(struct ldb_kv_private *ldb_kv,
+				struct dn_list *list);
+
+/* we put a @IDXVERSION attribute on index entries. This
+   allows us to tell if it was written by an older version
+*/
+#define LDB_KV_INDEXING_VERSION 2
+
+#define LDB_KV_GUID_INDEXING_VERSION 3
+
+static unsigned ldb_kv_max_key_length(struct ldb_kv_private *ldb_kv)
+{
+	if (ldb_kv->max_key_length == 0) {
+		return UINT_MAX;
+	}
+	return ldb_kv->max_key_length;
+}
+
+/* enable the idxptr mode when transactions start */
+int ldb_kv_index_transaction_start(
+	struct ldb_module *module,
+	size_t cache_size)
+{
+	struct ldb_kv_private *ldb_kv = talloc_get_type(
+	    ldb_module_get_private(module), struct ldb_kv_private);
+	ldb_kv->idxptr = talloc_zero(ldb_kv, struct ldb_kv_idxptr);
+	if (ldb_kv->idxptr == NULL) {
+		return ldb_oom(ldb_module_get_ctx(module));
+	}
+
+	ldb_kv->idxptr->itdb = tdb_open(
+		NULL,
+		cache_size,
+		TDB_INTERNAL,
+		O_RDWR,
+		0);
+	if (ldb_kv->idxptr->itdb == NULL) {
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	return LDB_SUCCESS;
+}
+
+/*
+  see if two ldb_val structures contain exactly the same data
+  return -1 or 1 for a mismatch, 0 for match
+*/
+static int ldb_val_equal_exact_for_qsort(const struct ldb_val *v1,
+					 const struct ldb_val *v2)
+{
+	if (v1->length > v2->length) {
+		return -1;
+	}
+	if (v1->length < v2->length) {
+		return 1;
+	}
+	return memcmp(v1->data, v2->data, v1->length);
+}
+
+/*
+  see if two ldb_val structures contain exactly the same data
+  return -1 or 1 for a mismatch, 0 for match
+*/
+static int ldb_val_equal_exact_ordered(const struct ldb_val v1,
+				       const struct ldb_val *v2)
+{
+	if (v1.length > v2->length) {
+		return -1;
+	}
+	if (v1.length < v2->length) {
+		return 1;
+	}
+	return memcmp(v1.data, v2->data, v1.length);
+}
+
+
+/*
+  find a entry in a dn_list, using a ldb_val. Uses a case sensitive
+  binary-safe comparison for the 'dn' returns -1 if not found
+
+  This is therefore safe when the value is a GUID in the future
+ */
+static int ldb_kv_dn_list_find_val(struct ldb_kv_private *ldb_kv,
+				   const struct dn_list *list,
+				   const struct ldb_val *v)
+{
+	unsigned int i;
+	struct ldb_val *exact = NULL, *next = NULL;
+
+	if (ldb_kv->cache->GUID_index_attribute == NULL) {
+		for (i=0; i<list->count; i++) {
+			if (ldb_val_equal_exact(&list->dn[i], v) == 1) {
+				return i;
+			}
+		}
+		return -1;
+	}
+
+	BINARY_ARRAY_SEARCH_GTE(list->dn, list->count,
+				*v, ldb_val_equal_exact_ordered,
+				exact, next);
+	if (exact == NULL) {
+		return -1;
+	}
+	/* Not required, but keeps the compiler quiet */
+	if (next != NULL) {
+		return -1;
+	}
+
+	i = exact - list->dn;
+	return i;
+}
+
+/*
+  find a entry in a dn_list. Uses a case sensitive comparison with the dn
+  returns -1 if not found
+ */
+static int ldb_kv_dn_list_find_msg(struct ldb_kv_private *ldb_kv,
+				   struct dn_list *list,
+				   const struct ldb_message *msg)
+{
+	struct ldb_val v;
+	const struct ldb_val *key_val;
+	if (ldb_kv->cache->GUID_index_attribute == NULL) {
+		const char *dn_str = ldb_dn_get_linearized(msg->dn);
+		v.data = discard_const_p(unsigned char, dn_str);
+		v.length = strlen(dn_str);
+	} else {
+		key_val = ldb_msg_find_ldb_val(
+		    msg, ldb_kv->cache->GUID_index_attribute);
+		if (key_val == NULL) {
+			return -1;
+		}
+		v = *key_val;
+	}
+	return ldb_kv_dn_list_find_val(ldb_kv, list, &v);
+}
+
+/*
+  this is effectively a cast function, but with lots of paranoia
+  checks and also copes with CPUs that are fussy about pointer
+  alignment
+ */
+static struct dn_list *ldb_kv_index_idxptr(struct ldb_module *module,
+					   TDB_DATA rec)
+{
+	struct dn_list *list;
+	if (rec.dsize != sizeof(void *)) {
+		ldb_asprintf_errstring(ldb_module_get_ctx(module),
+				       "Bad data size for idxptr %u", (unsigned)rec.dsize);
+		return NULL;
+	}
+	/* note that we can't just use a cast here, as rec.dptr may
+	   not be aligned sufficiently for a pointer. A cast would cause
+	   platforms like some ARM CPUs to crash */
+	memcpy(&list, rec.dptr, sizeof(void *));
+	list = talloc_get_type(list, struct dn_list);
+	if (list == NULL) {
+		ldb_asprintf_errstring(ldb_module_get_ctx(module),
+				       "Bad type '%s' for idxptr",
+				       talloc_get_name(list));
+		return NULL;
+	}
+	return list;
+}
+
+enum dn_list_will_be_read_only {
+	DN_LIST_MUTABLE = 0,
+	DN_LIST_WILL_BE_READ_ONLY = 1,
+};
+
+/*
+  return the @IDX list in an index entry for a dn as a
+  struct dn_list
+ */
+static int ldb_kv_dn_list_load(struct ldb_module *module,
+			       struct ldb_kv_private *ldb_kv,
+			       struct ldb_dn *dn,
+			       struct dn_list *list,
+			       enum dn_list_will_be_read_only read_only)
+{
+	struct ldb_message *msg;
+	int ret, version;
+	struct ldb_message_element *el;
+	TDB_DATA rec = {0};
+	struct dn_list *list2;
+	bool from_primary_cache = false;
+	TDB_DATA key = {0};
+
+	list->dn = NULL;
+	list->count = 0;
+	list->strict = false;
+
+	/*
+	 * See if we have an in memory index cache
+	 */
+	if (ldb_kv->idxptr == NULL) {
+		goto normal_index;
+	}
+
+	key.dptr = discard_const_p(unsigned char, ldb_dn_get_linearized(dn));
+	key.dsize = strlen((char *)key.dptr);
+
+	/*
+	 * Have we cached this index record?
+	 * If we have a nested transaction cache try that first.
+	 * then try the transaction cache.
+	 * if the record is not cached it will need to be read from disk.
+	 */
+	if (ldb_kv->nested_idx_ptr != NULL) {
+		rec = tdb_fetch(ldb_kv->nested_idx_ptr->itdb, key);
+	}
+	if (rec.dptr == NULL) {
+		from_primary_cache = true;
+		rec = tdb_fetch(ldb_kv->idxptr->itdb, key);
+	}
+	if (rec.dptr == NULL) {
+		goto normal_index;
+	}
+
+	/* we've found an in-memory index entry */
+	list2 = ldb_kv_index_idxptr(module, rec);
+	if (list2 == NULL) {
+		free(rec.dptr);
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+	free(rec.dptr);
+
+	/*
+	 * If this is a read only transaction the indexes will not be
+	 * changed so we don't need a copy in the event of a rollback
+	 *
+	 * In this case make an early return
+	 */
+	if (read_only == DN_LIST_WILL_BE_READ_ONLY) {
+		*list = *list2;
+		return LDB_SUCCESS;
+	}
+
+	/*
+	 * record was read from the sub transaction cache, so we have
+	 * already copied the primary cache record
+	 */
+	if (!from_primary_cache) {
+		*list = *list2;
+		return LDB_SUCCESS;
+	}
+
+	/*
+	 * No index sub transaction active, so no need to cache a copy
+	 */
+	if (ldb_kv->nested_idx_ptr == NULL) {
+		*list = *list2;
+		return LDB_SUCCESS;
+	}
+
+	/*
+	 * There is an active index sub transaction, and the record was
+	 * found in the primary index transaction cache.  A copy of the
+	 * record needs be taken to prevent the original entry being
+	 * altered, until the index sub transaction is committed.
+	 */
+
+	{
+		struct ldb_val *dns = NULL;
+		size_t x = 0;
+
+		dns = talloc_array(
+			list,
+			struct ldb_val,
+			list2->count);
+		if (dns == NULL) {
+			return LDB_ERR_OPERATIONS_ERROR;
+		}
+		for (x = 0; x < list2->count; x++) {
+			dns[x].length = list2->dn[x].length;
+			dns[x].data = talloc_memdup(
+				dns,
+				list2->dn[x].data,
+				list2->dn[x].length);
+			if (dns[x].data == NULL) {
+				TALLOC_FREE(dns);
+				return LDB_ERR_OPERATIONS_ERROR;
+			}
+		}
+		list->dn = dns;
+		list->count = list2->count;
+	}
+	return LDB_SUCCESS;
+
+	/*
+	 * Index record not found in the caches, read it from the
+	 * database.
+	 */
+normal_index:
+	msg = ldb_msg_new(list);
+	if (msg == NULL) {
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	ret = ldb_kv_search_dn1(module,
+				dn,
+				msg,
+				LDB_UNPACK_DATA_FLAG_NO_DN |
+				/*
+				 * The entry point ldb_kv_search_indexed is
+				 * only called from the read-locked
+				 * ldb_kv_search.
+				 */
+				LDB_UNPACK_DATA_FLAG_READ_LOCKED);
+	if (ret != LDB_SUCCESS) {
+		talloc_free(msg);
+		return ret;
+	}
+
+	el = ldb_msg_find_element(msg, LDB_KV_IDX);
+	if (!el) {
+		talloc_free(msg);
+		return LDB_SUCCESS;
+	}
+
+	version = ldb_msg_find_attr_as_int(msg, LDB_KV_IDXVERSION, 0);
+
+	/*
+	 * we avoid copying the strings by stealing the list.  We have
+	 * to steal msg onto el->values (which looks odd) because
+	 * the memory is allocated on msg, not on each value.
+	 */
+	if (ldb_kv->cache->GUID_index_attribute == NULL) {
+		/* check indexing version number */
+		if (version != LDB_KV_INDEXING_VERSION) {
+			ldb_debug_set(ldb_module_get_ctx(module),
+				      LDB_DEBUG_ERROR,
+				      "Wrong DN index version %d "
+				      "expected %d for %s",
+				      version, LDB_KV_INDEXING_VERSION,
+				      ldb_dn_get_linearized(dn));
+			talloc_free(msg);
+			return LDB_ERR_OPERATIONS_ERROR;
+		}
+
+		talloc_steal(el->values, msg);
+		list->dn = talloc_steal(list, el->values);
+		list->count = el->num_values;
+	} else {
+		unsigned int i;
+		if (version != LDB_KV_GUID_INDEXING_VERSION) {
+			/* This is quite likely during the DB startup
+			   on first upgrade to using a GUID index */
+			ldb_debug_set(ldb_module_get_ctx(module),
+				      LDB_DEBUG_ERROR,
+				      "Wrong GUID index version %d "
+				      "expected %d for %s",
+				      version, LDB_KV_GUID_INDEXING_VERSION,
+				      ldb_dn_get_linearized(dn));
+			talloc_free(msg);
+			return LDB_ERR_OPERATIONS_ERROR;
+		}
+
+		if (el->num_values == 0) {
+			talloc_free(msg);
+			return LDB_ERR_OPERATIONS_ERROR;
+		}
+
+		if ((el->values[0].length % LDB_KV_GUID_SIZE) != 0) {
+			talloc_free(msg);
+			return LDB_ERR_OPERATIONS_ERROR;
+		}
+
+		list->count = el->values[0].length / LDB_KV_GUID_SIZE;
+		list->dn = talloc_array(list, struct ldb_val, list->count);
+		if (list->dn == NULL) {
+			talloc_free(msg);
+			return LDB_ERR_OPERATIONS_ERROR;
+		}
+
+		/*
+		 * The actual data is on msg.
+		 */
+		talloc_steal(list->dn, msg);
+		for (i = 0; i < list->count; i++) {
+			list->dn[i].data
+				= &el->values[0].data[i * LDB_KV_GUID_SIZE];
+			list->dn[i].length = LDB_KV_GUID_SIZE;
+		}
+	}
+
+	/* We don't need msg->elements any more */
+	talloc_free(msg->elements);
+	return LDB_SUCCESS;
+}
+
+int ldb_kv_key_dn_from_idx(struct ldb_module *module,
+			   struct ldb_kv_private *ldb_kv,
+			   TALLOC_CTX *mem_ctx,
+			   struct ldb_dn *dn,
+			   struct ldb_val *ldb_key)
+{
+	struct ldb_context *ldb = ldb_module_get_ctx(module);
+	int ret;
+	int index = 0;
+	enum key_truncation truncation = KEY_NOT_TRUNCATED;
+	struct dn_list *list = talloc(mem_ctx, struct dn_list);
+	if (list == NULL) {
+		ldb_oom(ldb);
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	ret = ldb_kv_index_dn_base_dn(module, ldb_kv, dn, list, &truncation);
+	if (ret != LDB_SUCCESS) {
+		TALLOC_FREE(list);
+		return ret;
+	}
+
+	if (list->count == 0) {
+		TALLOC_FREE(list);
+		return LDB_ERR_NO_SUCH_OBJECT;
+	}
+
+	if (list->count > 1 && truncation == KEY_NOT_TRUNCATED)  {
+		const char *dn_str = ldb_dn_get_linearized(dn);
+		ldb_asprintf_errstring(ldb_module_get_ctx(module),
+				       __location__
+				       ": Failed to read DN index "
+				       "against %s for %s: too many "
+				       "values (%u > 1)",
+				       ldb_kv->cache->GUID_index_attribute,
+				       dn_str,
+				       list->count);
+		TALLOC_FREE(list);
+		return LDB_ERR_CONSTRAINT_VIOLATION;
+	}
+
+	if (list->count > 0 && truncation == KEY_TRUNCATED)  {
+		/*
+		 * DN key has been truncated, need to inspect the actual
+		 * records to locate the actual DN
+		 */
+		unsigned int i;
+		index = -1;
+		for (i=0; i < list->count; i++) {
+			uint8_t guid_key[LDB_KV_GUID_KEY_SIZE];
+			struct ldb_val key = {
+				.data = guid_key,
+				.length = sizeof(guid_key)
+			};
+			const int flags = LDB_UNPACK_DATA_FLAG_NO_ATTRS;
+			struct ldb_message *rec = ldb_msg_new(ldb);
+			if (rec == NULL) {
+				TALLOC_FREE(list);
+				return LDB_ERR_OPERATIONS_ERROR;
+			}
+
+			ret = ldb_kv_idx_to_key(
+			    module, ldb_kv, ldb, &list->dn[i], &key);
+			if (ret != LDB_SUCCESS) {
+				TALLOC_FREE(list);
+				TALLOC_FREE(rec);
+				return ret;
+			}
+
+			ret =
+			    ldb_kv_search_key(module, ldb_kv, key, rec, flags);
+			if (key.data != guid_key) {
+				TALLOC_FREE(key.data);
+			}
+			if (ret == LDB_ERR_NO_SUCH_OBJECT) {
+				/*
+				 * the record has disappeared?
+				 * yes, this can happen
+				 */
+				TALLOC_FREE(rec);
+				continue;
+			}
+
+			if (ret != LDB_SUCCESS) {
+				/* an internal error */
+				TALLOC_FREE(rec);
+				TALLOC_FREE(list);
+				return LDB_ERR_OPERATIONS_ERROR;
+			}
+
+			/*
+			 * We found the actual DN that we wanted from in the
+			 * multiple values that matched the index
+			 * (due to truncation), so return that.
+			 *
+			 */
+			if (ldb_dn_compare(dn, rec->dn) == 0) {
+				index = i;
+				TALLOC_FREE(rec);
+				break;
+			}
+		}
+
+		/*
+		 * We matched the index but the actual DN we wanted
+		 * was not here.
+		 */
+		if (index == -1) {
+			TALLOC_FREE(list);
+			return LDB_ERR_NO_SUCH_OBJECT;
+		}
+	}
+
+	/* The ldb_key memory is allocated by the caller */
+	ret = ldb_kv_guid_to_key(&list->dn[index], ldb_key);
+	TALLOC_FREE(list);
+
+	if (ret != LDB_SUCCESS) {
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	return LDB_SUCCESS;
+}
+
+
+
+/*
+  save a dn_list into a full @IDX style record
+ */
+static int ldb_kv_dn_list_store_full(struct ldb_module *module,
+				     struct ldb_kv_private *ldb_kv,
+				     struct ldb_dn *dn,
+				     struct dn_list *list)
+{
+	struct ldb_message *msg;
+	int ret;
+
+	msg = ldb_msg_new(module);
+	if (!msg) {
+		return ldb_module_oom(module);
+	}
+
+	msg->dn = dn;
+
+	if (list->count == 0) {
+		ret = ldb_kv_delete_noindex(module, msg);
+		if (ret == LDB_ERR_NO_SUCH_OBJECT) {
+			ret = LDB_SUCCESS;
+		}
+		TALLOC_FREE(msg);
+		return ret;
+	}
+
+	if (ldb_kv->cache->GUID_index_attribute == NULL) {
+		ret = ldb_msg_add_fmt(msg, LDB_KV_IDXVERSION, "%u",
+				      LDB_KV_INDEXING_VERSION);
+		if (ret != LDB_SUCCESS) {
+			TALLOC_FREE(msg);
+			return ldb_module_oom(module);
+		}
+	} else {
+		ret = ldb_msg_add_fmt(msg, LDB_KV_IDXVERSION, "%u",
+				      LDB_KV_GUID_INDEXING_VERSION);
+		if (ret != LDB_SUCCESS) {
+			TALLOC_FREE(msg);
+			return ldb_module_oom(module);
+		}
+	}
+
+	if (list->count > 0) {
+		struct ldb_message_element *el;
+
+		ret = ldb_msg_add_empty(msg, LDB_KV_IDX, LDB_FLAG_MOD_ADD, &el);
+		if (ret != LDB_SUCCESS) {
+			TALLOC_FREE(msg);
+			return ldb_module_oom(module);
+		}
+
+		if (ldb_kv->cache->GUID_index_attribute == NULL) {
+			el->values = list->dn;
+			el->num_values = list->count;
+		} else {
+			struct ldb_val v;
+			unsigned int i;
+			el->values = talloc_array(msg,
+						  struct ldb_val, 1);
+			if (el->values == NULL) {
+				TALLOC_FREE(msg);
+				return ldb_module_oom(module);
+			}
+
+			v.data = talloc_array_size(el->values,
+						   list->count,
+						   LDB_KV_GUID_SIZE);
+			if (v.data == NULL) {
+				TALLOC_FREE(msg);
+				return ldb_module_oom(module);
+			}
+
+			v.length = talloc_get_size(v.data);
+
+			for (i = 0; i < list->count; i++) {
+				if (list->dn[i].length !=
+				    LDB_KV_GUID_SIZE) {
+					TALLOC_FREE(msg);
+					return ldb_module_operr(module);
+				}
+				memcpy(&v.data[LDB_KV_GUID_SIZE*i],
+				       list->dn[i].data,
+				       LDB_KV_GUID_SIZE);
+			}
+			el->values[0] = v;
+			el->num_values = 1;
+		}
+	}
+
+	ret = ldb_kv_store(module, msg, TDB_REPLACE);
+	TALLOC_FREE(msg);
+	return ret;
+}
+
+/*
+  save a dn_list into the database, in either @IDX or internal format
+ */
+static int ldb_kv_dn_list_store(struct ldb_module *module,
+				struct ldb_dn *dn,
+				struct dn_list *list)
+{
+	struct ldb_kv_private *ldb_kv = talloc_get_type(
+	    ldb_module_get_private(module), struct ldb_kv_private);
+	TDB_DATA rec = {0};
+	TDB_DATA key = {0};
+
+	int ret = LDB_SUCCESS;
+	struct dn_list *list2 = NULL;
+	struct ldb_kv_idxptr *idxptr = NULL;
+
+	key.dptr = discard_const_p(unsigned char, ldb_dn_get_linearized(dn));
+	if (key.dptr == NULL) {
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+	key.dsize = strlen((char *)key.dptr);
+
+	/*
+	 * If there is an index sub transaction active, update the
+	 * sub transaction index cache.  Otherwise update the
+	 * primary index cache
+	 */
+	if (ldb_kv->nested_idx_ptr != NULL) {
+		idxptr = ldb_kv->nested_idx_ptr;
+	} else {
+		idxptr = ldb_kv->idxptr;
+	}
+	/*
+	 * Get the cache entry for the index
+	 *
+	 * As the value in the cache is a pointer to a dn_list we update
+	 * the dn_list directly.
+	 *
+	 */
+	rec = tdb_fetch(idxptr->itdb, key);
+	if (rec.dptr != NULL) {
+		list2 = ldb_kv_index_idxptr(module, rec);
+		if (list2 == NULL) {
+			free(rec.dptr);
+			return LDB_ERR_OPERATIONS_ERROR;
+		}
+		free(rec.dptr);
+		/* Now put the updated pointer back in the cache */
+		if (list->dn == NULL) {
+			list2->dn = NULL;
+			list2->count = 0;
+		} else {
+			list2->dn = talloc_steal(list2, list->dn);
+			list2->count = list->count;
+		}
+		return LDB_SUCCESS;
+	}
+
+	list2 = talloc(idxptr, struct dn_list);
+	if (list2 == NULL) {
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+	list2->dn = talloc_steal(list2, list->dn);
+	list2->count = list->count;
+
+	rec.dptr = (uint8_t *)&list2;
+	rec.dsize = sizeof(void *);
+
+
+	/*
+	 * This is not a store into the main DB, but into an in-memory
+	 * TDB, so we don't need a guard on ltdb->read_only
+	 *
+	 * Also as we directly update the in memory dn_list for existing
+	 * cache entries we must be adding a new entry to the cache.
+	 */
+	ret = tdb_store(idxptr->itdb, key, rec, TDB_INSERT);
+	if (ret != 0) {
+		return ltdb_err_map( tdb_error(idxptr->itdb));
+	}
+	return LDB_SUCCESS;
+}
+
+/*
+  traverse function for storing the in-memory index entries on disk
+ */
+static int ldb_kv_index_traverse_store(_UNUSED_ struct tdb_context *tdb,
+				       TDB_DATA key,
+				       TDB_DATA data,
+				       void *state)
+{
+	struct ldb_module *module = state;
+	struct ldb_kv_private *ldb_kv = talloc_get_type(
+	    ldb_module_get_private(module), struct ldb_kv_private);
+	struct ldb_dn *dn;
+	struct ldb_context *ldb = ldb_module_get_ctx(module);
+	struct ldb_val v;
+	struct dn_list *list;
+
+	list = ldb_kv_index_idxptr(module, data);
+	if (list == NULL) {
+		ldb_kv->idxptr->error = LDB_ERR_OPERATIONS_ERROR;
+		return -1;
+	}
+
+	v.data = key.dptr;
+	v.length = strnlen((char *)key.dptr, key.dsize);
+
+	dn = ldb_dn_from_ldb_val(module, ldb, &v);
+	if (dn == NULL) {
+		ldb_asprintf_errstring(ldb, "Failed to parse index key %*.*s as an LDB DN", (int)v.length, (int)v.length, (const char *)v.data);
+		ldb_kv->idxptr->error = LDB_ERR_OPERATIONS_ERROR;
+		return -1;
+	}
+
+	ldb_kv->idxptr->error =
+	    ldb_kv_dn_list_store_full(module, ldb_kv, dn, list);
+	talloc_free(dn);
+	if (ldb_kv->idxptr->error != 0) {
+		return -1;
+	}
+	return 0;
+}
+
+/* cleanup the idxptr mode when transaction commits */
+int ldb_kv_index_transaction_commit(struct ldb_module *module)
+{
+	struct ldb_kv_private *ldb_kv = talloc_get_type(
+	    ldb_module_get_private(module), struct ldb_kv_private);
+	int ret;
+
+	struct ldb_context *ldb = ldb_module_get_ctx(module);
+
+	ldb_reset_err_string(ldb);
+
+	if (ldb_kv->idxptr->itdb) {
+		tdb_traverse(
+		    ldb_kv->idxptr->itdb, ldb_kv_index_traverse_store, module);
+		tdb_close(ldb_kv->idxptr->itdb);
+	}
+
+	ret = ldb_kv->idxptr->error;
+	if (ret != LDB_SUCCESS) {
+		if (!ldb_errstring(ldb)) {
+			ldb_set_errstring(ldb, ldb_strerror(ret));
+		}
+		ldb_asprintf_errstring(ldb, "Failed to store index records in transaction commit: %s", ldb_errstring(ldb));
+	}
+
+	talloc_free(ldb_kv->idxptr);
+	ldb_kv->idxptr = NULL;
+	return ret;
+}
+
+/* cleanup the idxptr mode when transaction cancels */
+int ldb_kv_index_transaction_cancel(struct ldb_module *module)
+{
+	struct ldb_kv_private *ldb_kv = talloc_get_type(
+	    ldb_module_get_private(module), struct ldb_kv_private);
+	if (ldb_kv->idxptr && ldb_kv->idxptr->itdb) {
+		tdb_close(ldb_kv->idxptr->itdb);
+	}
+	TALLOC_FREE(ldb_kv->idxptr);
+	if (ldb_kv->nested_idx_ptr && ldb_kv->nested_idx_ptr->itdb) {
+		tdb_close(ldb_kv->nested_idx_ptr->itdb);
+	}
+	TALLOC_FREE(ldb_kv->nested_idx_ptr);
+	return LDB_SUCCESS;
+}
+
+
+/*
+  return the dn key to be used for an index
+  the caller is responsible for freeing
+*/
+static struct ldb_dn *ldb_kv_index_key(struct ldb_context *ldb,
+				       TALLOC_CTX *mem_ctx,
+				       struct ldb_kv_private *ldb_kv,
+				       const char *attr,
+				       const struct ldb_val *value,
+				       const struct ldb_schema_attribute **ap,
+				       enum key_truncation *truncation)
+{
+	struct ldb_dn *ret;
+	struct ldb_val v;
+	const struct ldb_schema_attribute *a = NULL;
+	char *attr_folded = NULL;
+	const char *attr_for_dn = NULL;
+	int r;
+	bool should_b64_encode;
+
+	unsigned int max_key_length = ldb_kv_max_key_length(ldb_kv);
+	size_t key_len = 0;
+	size_t attr_len = 0;
+	const size_t indx_len = sizeof(LDB_KV_INDEX) - 1;
+	unsigned frmt_len = 0;
+	const size_t additional_key_length = 4;
+	unsigned int num_separators = 3; /* Estimate for overflow check */
+	const size_t min_data = 1;
+	const size_t min_key_length = additional_key_length
+		+ indx_len + num_separators + min_data;
+	struct ldb_val empty;
+
+	/*
+	 * Accept a NULL value as a request for a key with no value.  This is
+	 * different from passing an empty value, which might be given
+	 * significance by some canonicalise functions.
+	 */
+	bool empty_val = value == NULL;
+	if (empty_val) {
+		empty.length = 0;
+		empty.data = discard_const_p(unsigned char, "");
+		value = &empty;
+	}
+
+	if (attr[0] == '@') {
+		attr_for_dn = attr;
+		v = *value;
+		if (ap != NULL) {
+			*ap = NULL;
+		}
+	} else {
+		attr_folded = ldb_attr_casefold(ldb, attr);
+		if (!attr_folded) {
+			return NULL;
+		}
+
+		attr_for_dn = attr_folded;
+
+		a = ldb_schema_attribute_by_name(ldb, attr);
+		if (ap) {
+			*ap = a;
+		}
+
+		if (empty_val) {
+			v = *value;
+		} else {
+			ldb_attr_handler_t fn;
+			if (a->syntax->index_format_fn &&
+			    ldb_kv->cache->GUID_index_attribute != NULL) {
+				fn = a->syntax->index_format_fn;
+			} else {
+				fn = a->syntax->canonicalise_fn;
+			}
+			r = fn(ldb, ldb, value, &v);
+			if (r != LDB_SUCCESS) {
+				const char *errstr = ldb_errstring(ldb);
+				/* canonicalisation can be refused. For
+				   example, a attribute that takes wildcards
+				   will refuse to canonicalise 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);
+				return NULL;
+			}
+		}
+	}
+	attr_len = strlen(attr_for_dn);
+
+	/*
+	 * Check if there is any hope this will fit into the DB.
+	 * Overflow here is not actually critical the code below
+	 * checks again to make the printf and the DB does another
+	 * check for too long keys
+	 */
+	if (max_key_length - attr_len < min_key_length) {
+		ldb_asprintf_errstring(
+			ldb,
+			__location__ ": max_key_length "
+			"is too small (%u) < (%u)",
+			max_key_length,
+			(unsigned)(min_key_length + attr_len));
+		talloc_free(attr_folded);
+		return NULL;
+	}
+
+	/*
+	 * ltdb_key_dn() makes something 4 bytes longer, it adds a leading
+	 * "DN=" and a trailing string terminator
+	 */
+	max_key_length -= additional_key_length;
+
+	/*
+	 * We do not base 64 encode a DN in a key, it has already been
+	 * casefolded and linearized, that is good enough.  That already
+	 * avoids embedded NUL etc.
+	 */
+	if (ldb_kv->cache->GUID_index_attribute != NULL) {
+		if (strcmp(attr, LDB_KV_IDXDN) == 0) {
+			should_b64_encode = false;
+		} else if (strcmp(attr, LDB_KV_IDXONE) == 0) {
+			/*
+			 * We can only change the behaviour for IDXONE
+			 * when the GUID index is enabled
+			 */
+			should_b64_encode = false;
+		} else {
+			should_b64_encode
+				= ldb_should_b64_encode(ldb, &v);
+		}
+	} else {
+		should_b64_encode = ldb_should_b64_encode(ldb, &v);
+	}
+
+	if (should_b64_encode) {
+		size_t vstr_len = 0;
+		char *vstr = ldb_base64_encode(mem_ctx, (char *)v.data, v.length);
+		if (!vstr) {
+			talloc_free(attr_folded);
+			return NULL;
+		}
+		vstr_len = strlen(vstr);
+		/*
+		 * Overflow here is not critical as we only use this
+		 * to choose the printf truncation
+		 */
+		key_len = num_separators + indx_len + attr_len + vstr_len;
+		if (key_len > max_key_length) {
+			size_t excess = key_len - max_key_length;
+			frmt_len = vstr_len - excess;
+			*truncation = KEY_TRUNCATED;
+			/*
+			* Truncated keys are placed in a separate key space
+			* from the non truncated keys
+			* Note: the double hash "##" is not a typo and
+			* indicates that the following value is base64 encoded
+			*/
+			ret = ldb_dn_new_fmt(mem_ctx, ldb, "%s#%s##%.*s",
+					     LDB_KV_INDEX, attr_for_dn,
+					     frmt_len, vstr);
+		} else {
+			frmt_len = vstr_len;
+			*truncation = KEY_NOT_TRUNCATED;
+			/*
+			 * Note: the double colon "::" is not a typo and
+			 * indicates that the following value is base64 encoded
+			 */
+			ret = ldb_dn_new_fmt(mem_ctx, ldb, "%s:%s::%.*s",
+					     LDB_KV_INDEX, attr_for_dn,
+					     frmt_len, vstr);
+		}
+		talloc_free(vstr);
+	} else {
+		/* Only need two separators */
+		num_separators = 2;
+
+		/*
+		 * Overflow here is not critical as we only use this
+		 * to choose the printf truncation
+		 */
+		key_len = num_separators + indx_len + attr_len + (int)v.length;
+		if (key_len > max_key_length) {
+			size_t excess = key_len - max_key_length;
+			frmt_len = v.length - excess;
+			*truncation = KEY_TRUNCATED;
+			/*
+			 * Truncated keys are placed in a separate key space
+			 * from the non truncated keys
+			 */
+			ret = ldb_dn_new_fmt(mem_ctx, ldb, "%s#%s#%.*s",
+					     LDB_KV_INDEX, attr_for_dn,
+					     frmt_len, (char *)v.data);
+		} else {
+			frmt_len = v.length;
+			*truncation = KEY_NOT_TRUNCATED;
+			ret = ldb_dn_new_fmt(mem_ctx, ldb, "%s:%s:%.*s",
+					     LDB_KV_INDEX, attr_for_dn,
+					     frmt_len, (char *)v.data);
+		}
+	}
+
+	if (value != NULL && v.data != value->data && !empty_val) {
+		talloc_free(v.data);
+	}
+	talloc_free(attr_folded);
+
+	return ret;
+}
+
+/*
+  see if a attribute value is in the list of indexed attributes
+*/
+static bool ldb_kv_is_indexed(struct ldb_module *module,
+			      struct ldb_kv_private *ldb_kv,
+			      const char *attr)
+{
+	struct ldb_context *ldb = ldb_module_get_ctx(module);
+	unsigned int i;
+	struct ldb_message_element *el;
+
+	if ((ldb_kv->cache->GUID_index_attribute != NULL) &&
+	    (ldb_attr_cmp(attr, ldb_kv->cache->GUID_index_attribute) == 0)) {
+		/* Implicitly covered, this is the index key */
+		return false;
+	}
+	if (ldb->schema.index_handler_override) {
+		const struct ldb_schema_attribute *a
+			= ldb_schema_attribute_by_name(ldb, attr);
+
+		if (a == NULL) {
+			return false;
+		}
+
+		if (a->flags & LDB_ATTR_FLAG_INDEXED) {
+			return true;
+		} else {
+			return false;
+		}
+	}
+
+	if (!ldb_kv->cache->attribute_indexes) {
+		return false;
+	}
+
+	el = ldb_msg_find_element(ldb_kv->cache->indexlist, LDB_KV_IDXATTR);
+	if (el == NULL) {
+		return false;
+	}
+
+	/* TODO: this is too expensive! At least use a binary search */
+	for (i=0; i<el->num_values; i++) {
+		if (ldb_attr_cmp((char *)el->values[i].data, attr) == 0) {
+			return true;
+		}
+	}
+	return false;
+}
+
+/*
+  in the following logic functions, the return value is treated as
+  follows:
+
+     LDB_SUCCESS: we found some matching index values
+
+     LDB_ERR_NO_SUCH_OBJECT: we know for sure that no object matches
+
+     LDB_ERR_OPERATIONS_ERROR: indexing could not answer the call,
+                               we'll need a full search
+ */
+
+/*
+  return a list of dn's that might match a simple indexed search (an
+  equality search only)
+ */
+static int ldb_kv_index_dn_simple(struct ldb_module *module,
+				  struct ldb_kv_private *ldb_kv,
+				  const struct ldb_parse_tree *tree,
+				  struct dn_list *list)
+{
+	struct ldb_context *ldb;
+	struct ldb_dn *dn;
+	int ret;
+	enum key_truncation truncation = KEY_NOT_TRUNCATED;
+
+	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_kv_is_indexed(module, ldb_kv, tree->u.equality.attr)) {
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	/*
+	 * the attribute is indexed. Pull the list of DNs that match the
+	 * search criterion
+	 *
+	 * list is used as a memory context as it has a shorter life
+	 * than 'ldb'.  Regardless we talloc_free() 'dn' below.
+	 */
+	dn = ldb_kv_index_key(ldb,
+			      list,
+			      ldb_kv,
+			      tree->u.equality.attr,
+			      &tree->u.equality.value,
+			      NULL,
+			      &truncation);
+	/*
+	 * We ignore truncation here and allow multi-valued matches
+	 * as ltdb_search_indexed will filter out the wrong one in
+	 * ltdb_index_filter() which calls ldb_match_message().
+	 */
+	if (!dn) {
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	ret = ldb_kv_dn_list_load(module, ldb_kv, dn, list,
+				  DN_LIST_WILL_BE_READ_ONLY);
+	talloc_free(dn);
+	return ret;
+}
+
+static bool list_union(struct ldb_context *ldb,
+		       struct ldb_kv_private *ldb_kv,
+		       struct dn_list *list,
+		       struct dn_list *list2);
+
+/*
+  return a list of dn's that might match a leaf indexed search
+ */
+static int ldb_kv_index_dn_leaf(struct ldb_module *module,
+				struct ldb_kv_private *ldb_kv,
+				const struct ldb_parse_tree *tree,
+				struct dn_list *list)
+{
+	if (ldb_kv->disallow_dn_filter &&
+	    (ldb_attr_cmp(tree->u.equality.attr, "dn") == 0)) {
+		/* in AD mode we do not support "(dn=...)" search filters */
+		list->dn = NULL;
+		list->count = 0;
+		return LDB_SUCCESS;
+	}
+	if (tree->u.equality.attr[0] == '@') {
+		/* Do not allow a indexed search against an @ */
+		list->dn = NULL;
+		list->count = 0;
+		return LDB_SUCCESS;
+	}
+	if (ldb_attr_dn(tree->u.equality.attr) == 0) {
+		enum key_truncation truncation = KEY_NOT_TRUNCATED;
+		bool valid_dn = false;
+		struct ldb_dn *dn
+			= ldb_dn_from_ldb_val(list,
+					      ldb_module_get_ctx(module),
+					      &tree->u.equality.value);
+		if (dn == NULL) {
+			/* If we can't parse it, no match */
+			list->dn = NULL;
+			list->count = 0;
+			return LDB_SUCCESS;
+		}
+
+		valid_dn = ldb_dn_validate(dn);
+		if (valid_dn == false) {
+			/* If we can't parse it, no match */
+			list->dn = NULL;
+			list->count = 0;
+			return LDB_SUCCESS;
+		}
+
+		/*
+		 * Re-use the same code we use for a SCOPE_BASE
+		 * search
+		 *
+		 * We can't call TALLOC_FREE(dn) as this must belong
+		 * to list for the memory to remain valid.
+		 */
+		return ldb_kv_index_dn_base_dn(
+		    module, ldb_kv, dn, list, &truncation);
+		/*
+		 * We ignore truncation here and allow multi-valued matches
+		 * as ltdb_search_indexed will filter out the wrong one in
+		 * ltdb_index_filter() which calls ldb_match_message().
+		 */
+
+	} else if ((ldb_kv->cache->GUID_index_attribute != NULL) &&
+		   (ldb_attr_cmp(tree->u.equality.attr,
+				 ldb_kv->cache->GUID_index_attribute) == 0)) {
+		int ret;
+		struct ldb_context *ldb = ldb_module_get_ctx(module);
+		list->dn = talloc_array(list, struct ldb_val, 1);
+		if (list->dn == NULL) {
+			ldb_module_oom(module);
+			return LDB_ERR_OPERATIONS_ERROR;
+		}
+		/*
+		 * We need to go via the canonicalise_fn() to
+		 * ensure we get the index in binary, rather
+		 * than a string
+		 */
+		ret = ldb_kv->GUID_index_syntax->canonicalise_fn(
+		    ldb, list->dn, &tree->u.equality.value, &list->dn[0]);
+		if (ret != LDB_SUCCESS) {
+			return LDB_ERR_OPERATIONS_ERROR;
+		}
+		list->count = 1;
+		return LDB_SUCCESS;
+	}
+
+	return ldb_kv_index_dn_simple(module, ldb_kv, tree, list);
+}
+
+
+/*
+  list intersection
+  list = list & list2
+*/
+static bool list_intersect(struct ldb_kv_private *ldb_kv,
+			   struct dn_list *list,
+			   const struct dn_list *list2)
+{
+	const struct dn_list *short_list, *long_list;
+	struct dn_list *list3;
+	unsigned int i;
+
+	if (list->count == 0) {
+		/* 0 & X == 0 */
+		return true;
+	}
+	if (list2->count == 0) {
+		/* X & 0 == 0 */
+		list->count = 0;
+		list->dn = NULL;
+		return true;
+	}
+
+	/*
+	 * In both of the below we check for strict and in that
+	 * case do not optimise the intersection of this list,
+	 * we must never return an entry not in this
+	 * list.  This allows the index for
+	 * SCOPE_ONELEVEL to be trusted.
+	 */
+
+	/* the indexing code is allowed to return a longer list than
+	   what really matches, as all results are filtered by the
+	   full expression at the end - this shortcut avoids a lot of
+	   work in some cases */
+	if (list->count < 2 && list2->count > 10 && list2->strict == false) {
+		return true;
+	}
+	if (list2->count < 2 && list->count > 10 && list->strict == false) {
+		list->count = list2->count;
+		list->dn = list2->dn;
+		/* note that list2 may not be the parent of list2->dn,
+		   as list2->dn may be owned by ltdb->idxptr. In that
+		   case we expect this reparent call to fail, which is
+		   OK */
+		talloc_reparent(list2, list, list2->dn);
+		return true;
+	}
+
+	if (list->count > list2->count) {
+		short_list = list2;
+		long_list = list;
+	} else {
+		short_list = list;
+		long_list = list2;
+	}
+
+	list3 = talloc_zero(list, struct dn_list);
+	if (list3 == NULL) {
+		return false;
+	}
+
+	list3->dn = talloc_array(list3, struct ldb_val,
+				 MIN(list->count, list2->count));
+	if (!list3->dn) {
+		talloc_free(list3);
+		return false;
+	}
+	list3->count = 0;
+
+	for (i=0;i<short_list->count;i++) {
+		/* For the GUID index case, this is a binary search */
+		if (ldb_kv_dn_list_find_val(
+			ldb_kv, long_list, &short_list->dn[i]) != -1) {
+			list3->dn[list3->count] = short_list->dn[i];
+			list3->count++;
+		}
+	}
+
+	list->strict |= list2->strict;
+	list->dn = talloc_steal(list, list3->dn);
+	list->count = list3->count;
+	talloc_free(list3);
+
+	return true;
+}
+
+
+/*
+  list union
+  list = list | list2
+*/
+static bool list_union(struct ldb_context *ldb,
+		       struct ldb_kv_private *ldb_kv,
+		       struct dn_list *list,
+		       struct dn_list *list2)
+{
+	struct ldb_val *dn3;
+	unsigned int i = 0, j = 0, k = 0;
+
+	if (list2->count == 0) {
+		/* X | 0 == X */
+		return true;
+	}
+
+	if (list->count == 0) {
+		/* 0 | X == X */
+		list->count = list2->count;
+		list->dn = list2->dn;
+		/* note that list2 may not be the parent of list2->dn,
+		   as list2->dn may be owned by ltdb->idxptr. In that
+		   case we expect this reparent call to fail, which is
+		   OK */
+		talloc_reparent(list2, list, list2->dn);
+		return true;
+	}
+
+	/*
+	 * Sort the lists (if not in GUID DN mode) so we can do
+	 * the de-duplication during the merge
+	 *
+	 * NOTE: This can sort the in-memory index values, as list or
+	 * list2 might not be a copy!
+	 */
+	ldb_kv_dn_list_sort(ldb_kv, list);
+	ldb_kv_dn_list_sort(ldb_kv, list2);
+
+	dn3 = talloc_array(list, struct ldb_val, list->count + list2->count);
+	if (!dn3) {
+		ldb_oom(ldb);
+		return false;
+	}
+
+	while (i < list->count || j < list2->count) {
+		int cmp;
+		if (i >= list->count) {
+			cmp = 1;
+		} else if (j >= list2->count) {
+			cmp = -1;
+		} else {
+			cmp = ldb_val_equal_exact_ordered(list->dn[i],
+							  &list2->dn[j]);
+		}
+
+		if (cmp < 0) {
+			/* Take list */
+			dn3[k] = list->dn[i];
+			i++;
+			k++;
+		} else if (cmp > 0) {
+			/* Take list2 */
+			dn3[k] = list2->dn[j];
+			j++;
+			k++;
+		} else {
+			/* Equal, take list */
+			dn3[k] = list->dn[i];
+			i++;
+			j++;
+			k++;
+		}
+	}
+
+	list->dn = dn3;
+	list->count = k;
+
+	return true;
+}
+
+static int ldb_kv_index_dn(struct ldb_module *module,
+			   struct ldb_kv_private *ldb_kv,
+			   const struct ldb_parse_tree *tree,
+			   struct dn_list *list);
+
+/*
+  process an OR list (a union)
+ */
+static int ldb_kv_index_dn_or(struct ldb_module *module,
+			      struct ldb_kv_private *ldb_kv,
+			      const struct ldb_parse_tree *tree,
+			      struct dn_list *list)
+{
+	struct ldb_context *ldb;
+	unsigned int i;
+
+	ldb = ldb_module_get_ctx(module);
+
+	list->dn = NULL;
+	list->count = 0;
+
+	for (i=0; i<tree->u.list.num_elements; i++) {
+		struct dn_list *list2;
+		int ret;
+
+		list2 = talloc_zero(list, struct dn_list);
+		if (list2 == NULL) {
+			return LDB_ERR_OPERATIONS_ERROR;
+		}
+
+		ret = ldb_kv_index_dn(
+		    module, ldb_kv, tree->u.list.elements[i], list2);
+
+		if (ret == LDB_ERR_NO_SUCH_OBJECT) {
+			/* X || 0 == X */
+			talloc_free(list2);
+			continue;
+		}
+
+		if (ret != LDB_SUCCESS) {
+			/* X || * == * */
+			talloc_free(list2);
+			return ret;
+		}
+
+		if (!list_union(ldb, ldb_kv, list, list2)) {
+			talloc_free(list2);
+			return LDB_ERR_OPERATIONS_ERROR;
+		}
+	}
+
+	if (list->count == 0) {
+		return LDB_ERR_NO_SUCH_OBJECT;
+	}
+
+	return LDB_SUCCESS;
+}
+
+
+/*
+  NOT an index results
+ */
+static int ldb_kv_index_dn_not(_UNUSED_ struct ldb_module *module,
+			       _UNUSED_ struct ldb_kv_private *ldb_kv,
+			       _UNUSED_ const struct ldb_parse_tree *tree,
+			       _UNUSED_ struct dn_list *list)
+{
+	/* the only way to do an indexed not would be if we could
+	   negate the not via another not or if we knew the total
+	   number of database elements so we could know that the
+	   existing expression covered the whole database.
+
+	   instead, we just give up, and rely on a full index scan
+	   (unless an outer & manages to reduce the list)
+	*/
+	return LDB_ERR_OPERATIONS_ERROR;
+}
+
+/*
+ * These things are unique, so avoid a full scan if this is a search
+ * by GUID, DN or a unique attribute
+ */
+static bool ldb_kv_index_unique(struct ldb_context *ldb,
+				struct ldb_kv_private *ldb_kv,
+				const char *attr)
+{
+	const struct ldb_schema_attribute *a;
+	if (ldb_kv->cache->GUID_index_attribute != NULL) {
+		if (ldb_attr_cmp(attr, ldb_kv->cache->GUID_index_attribute) ==
+		    0) {
+			return true;
+		}
+	}
+	if (ldb_attr_dn(attr) == 0) {
+		return true;
+	}
+
+	a = ldb_schema_attribute_by_name(ldb, attr);
+	if (a->flags & LDB_ATTR_FLAG_UNIQUE_INDEX) {
+		return true;
+	}
+	return false;
+}
+
+/*
+  process an AND expression (intersection)
+ */
+static int ldb_kv_index_dn_and(struct ldb_module *module,
+			       struct ldb_kv_private *ldb_kv,
+			       const struct ldb_parse_tree *tree,
+			       struct dn_list *list)
+{
+	struct ldb_context *ldb;
+	unsigned int i;
+	bool found;
+
+	ldb = ldb_module_get_ctx(module);
+
+	list->dn = NULL;
+	list->count = 0;
+
+	/* in the first pass we only look for unique simple
+	   equality tests, in the hope of avoiding having to look
+	   at any others */
+	for (i=0; i<tree->u.list.num_elements; i++) {
+		const struct ldb_parse_tree *subtree = tree->u.list.elements[i];
+		int ret;
+
+		if (subtree->operation != LDB_OP_EQUALITY ||
+		    !ldb_kv_index_unique(
+			ldb, ldb_kv, subtree->u.equality.attr)) {
+			continue;
+		}
+
+		ret = ldb_kv_index_dn(module, ldb_kv, subtree, list);
+		if (ret == LDB_ERR_NO_SUCH_OBJECT) {
+			/* 0 && X == 0 */
+			return LDB_ERR_NO_SUCH_OBJECT;
+		}
+		if (ret == LDB_SUCCESS) {
+			/* a unique index match means we can
+			 * stop. Note that we don't care if we return
+			 * a few too many objects, due to later
+			 * filtering */
+			return LDB_SUCCESS;
+		}
+	}
+
+	/* now do a full intersection */
+	found = false;
+
+	for (i=0; i<tree->u.list.num_elements; i++) {
+		const struct ldb_parse_tree *subtree = tree->u.list.elements[i];
+		struct dn_list *list2;
+		int ret;
+
+		list2 = talloc_zero(list, struct dn_list);
+		if (list2 == NULL) {
+			return ldb_module_oom(module);
+		}
+
+		ret = ldb_kv_index_dn(module, ldb_kv, subtree, list2);
+
+		if (ret == LDB_ERR_NO_SUCH_OBJECT) {
+			/* X && 0 == 0 */
+			list->dn = NULL;
+			list->count = 0;
+			talloc_free(list2);
+			return LDB_ERR_NO_SUCH_OBJECT;
+		}
+
+		if (ret != LDB_SUCCESS) {
+			/* this didn't adding anything */
+			talloc_free(list2);
+			continue;
+		}
+
+		if (!found) {
+			talloc_reparent(list2, list, list->dn);
+			list->dn = list2->dn;
+			list->count = list2->count;
+			found = true;
+		} else if (!list_intersect(ldb_kv, list, list2)) {
+			talloc_free(list2);
+			return LDB_ERR_OPERATIONS_ERROR;
+		}
+
+		if (list->count == 0) {
+			list->dn = NULL;
+			return LDB_ERR_NO_SUCH_OBJECT;
+		}
+
+		if (list->count < 2) {
+			/* it isn't worth loading the next part of the tree */
+			return LDB_SUCCESS;
+		}
+	}
+
+	if (!found) {
+		/* none of the attributes were indexed */
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	return LDB_SUCCESS;
+}
+
+struct ldb_kv_ordered_index_context {
+	struct ldb_module *module;
+	int error;
+	struct dn_list *dn_list;
+};
+
+static int traverse_range_index(_UNUSED_ struct ldb_kv_private *ldb_kv,
+				_UNUSED_ struct ldb_val key,
+				struct ldb_val data,
+				void *state)
+{
+
+	struct ldb_context *ldb;
+	struct ldb_kv_ordered_index_context *ctx =
+	    (struct ldb_kv_ordered_index_context *)state;
+	struct ldb_module *module = ctx->module;
+	struct ldb_message_element *el = NULL;
+	struct ldb_message *msg = NULL;
+	int version;
+	size_t dn_array_size, additional_length;
+	unsigned int i;
+
+	ldb = ldb_module_get_ctx(module);
+
+	msg = ldb_msg_new(module);
+
+	ctx->error = ldb_unpack_data_flags(ldb, &data, msg,
+					   LDB_UNPACK_DATA_FLAG_NO_DN);
+
+	if (ctx->error != LDB_SUCCESS) {
+		talloc_free(msg);
+		return ctx->error;
+	}
+
+	el = ldb_msg_find_element(msg, LDB_KV_IDX);
+	if (!el) {
+		talloc_free(msg);
+		return LDB_SUCCESS;
+	}
+
+	version = ldb_msg_find_attr_as_int(msg, LDB_KV_IDXVERSION, 0);
+
+	/*
+	 * we avoid copying the strings by stealing the list.  We have
+	 * to steal msg onto el->values (which looks odd) because
+	 * the memory is allocated on msg, not on each value.
+	 */
+	if (version != LDB_KV_GUID_INDEXING_VERSION) {
+		/* This is quite likely during the DB startup
+		   on first upgrade to using a GUID index */
+		ldb_debug_set(ldb_module_get_ctx(module),
+			      LDB_DEBUG_ERROR, __location__
+			      ": Wrong GUID index version %d expected %d",
+			      version, LDB_KV_GUID_INDEXING_VERSION);
+		talloc_free(msg);
+		ctx->error = LDB_ERR_OPERATIONS_ERROR;
+		return ctx->error;
+	}
+
+	if (el->num_values == 0) {
+		talloc_free(msg);
+		ctx->error = LDB_ERR_OPERATIONS_ERROR;
+		return ctx->error;
+	}
+
+	if ((el->values[0].length % LDB_KV_GUID_SIZE) != 0
+	    || el->values[0].length == 0) {
+		talloc_free(msg);
+		ctx->error = LDB_ERR_OPERATIONS_ERROR;
+		return ctx->error;
+	}
+
+	dn_array_size = talloc_array_length(ctx->dn_list->dn);
+
+	additional_length = el->values[0].length / LDB_KV_GUID_SIZE;
+
+	if (ctx->dn_list->count + additional_length < ctx->dn_list->count) {
+		talloc_free(msg);
+		ctx->error = LDB_ERR_OPERATIONS_ERROR;
+		return ctx->error;
+	}
+
+	if ((ctx->dn_list->count + additional_length) >= dn_array_size) {
+		size_t new_array_length;
+
+		if (dn_array_size * 2 < dn_array_size) {
+			talloc_free(msg);
+			ctx->error = LDB_ERR_OPERATIONS_ERROR;
+			return ctx->error;
+		}
+
+		new_array_length = MAX(ctx->dn_list->count + additional_length,
+				       dn_array_size * 2);
+
+		ctx->dn_list->dn = talloc_realloc(ctx->dn_list,
+						  ctx->dn_list->dn,
+						  struct ldb_val,
+						  new_array_length);
+	}
+
+	if (ctx->dn_list->dn == NULL) {
+		talloc_free(msg);
+		ctx->error = LDB_ERR_OPERATIONS_ERROR;
+		return ctx->error;
+	}
+
+	/*
+	 * The actual data is on msg.
+	 */
+	talloc_steal(ctx->dn_list->dn, msg);
+	for (i = 0; i < additional_length; i++) {
+		ctx->dn_list->dn[i + ctx->dn_list->count].data
+			= &el->values[0].data[i * LDB_KV_GUID_SIZE];
+		ctx->dn_list->dn[i + ctx->dn_list->count].length = LDB_KV_GUID_SIZE;
+
+	}
+
+	ctx->dn_list->count += additional_length;
+
+	talloc_free(msg->elements);
+
+	return LDB_SUCCESS;
+}
+
+/*
+ * >= and <= indexing implemented using lexicographically sorted keys
+ *
+ * We only run this in GUID indexing mode and when there is no write
+ * transaction (only implicit read locks are being held). Otherwise, we would
+ * have to deal with the in-memory index cache.
+ *
+ * We rely on the implementation of index_format_fn on a schema syntax which
+ * will can help us to construct keys which can be ordered correctly, and we
+ * terminate using schema agnostic start and end keys.
+ *
+ * index_format_fn must output values which can be memcmp-able to produce the
+ * correct ordering as defined by the schema syntax class.
+ */
+static int ldb_kv_index_dn_ordered(struct ldb_module *module,
+				   struct ldb_kv_private *ldb_kv,
+				   const struct ldb_parse_tree *tree,
+				   struct dn_list *list, bool ascending)
+{
+	enum key_truncation truncation = KEY_NOT_TRUNCATED;
+	struct ldb_context *ldb = ldb_module_get_ctx(module);
+
+	struct ldb_val ldb_key = { 0 }, ldb_key2 = { 0 };
+	struct ldb_val start_key, end_key;
+	struct ldb_dn *key_dn = NULL;
+	const struct ldb_schema_attribute *a = NULL;
+
+	struct ldb_kv_ordered_index_context ctx;
+	int ret;
+
+	TALLOC_CTX *tmp_ctx = NULL;
+
+	if (!ldb_kv_is_indexed(module, ldb_kv, tree->u.comparison.attr)) {
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	if (ldb_kv->cache->GUID_index_attribute == NULL) {
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	/* bail out if we're in a transaction, full search instead. */
+	if (ldb_kv->kv_ops->transaction_active(ldb_kv)) {
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	if (ldb_kv->disallow_dn_filter &&
+	    (ldb_attr_cmp(tree->u.comparison.attr, "dn") == 0)) {
+		/* in AD mode we do not support "(dn=...)" search filters */
+		list->dn = NULL;
+		list->count = 0;
+		return LDB_SUCCESS;
+	}
+	if (tree->u.comparison.attr[0] == '@') {
+		/* Do not allow a indexed search against an @ */
+		list->dn = NULL;
+		list->count = 0;
+		return LDB_SUCCESS;
+	}
+
+	a = ldb_schema_attribute_by_name(ldb, tree->u.comparison.attr);
+
+	/*
+	 * If there's no index format function defined for this attr, then
+	 * the lexicographic order in the database doesn't correspond to the
+	 * attr's ordering, so we can't use the iterate_range op.
+	 */
+	if (a->syntax->index_format_fn == NULL) {
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	tmp_ctx = talloc_new(NULL);
+	if (tmp_ctx == NULL) {
+		return ldb_module_oom(module);
+	}
+
+	key_dn = ldb_kv_index_key(ldb, tmp_ctx, ldb_kv, tree->u.comparison.attr,
+				  &tree->u.comparison.value,
+				  NULL, &truncation);
+	if (!key_dn) {
+		TALLOC_FREE(tmp_ctx);
+		return LDB_ERR_OPERATIONS_ERROR;
+	} else if (truncation == KEY_TRUNCATED) {
+		ldb_debug(ldb, LDB_DEBUG_WARNING,
+			  __location__
+			  ": ordered index violation: key dn truncated: %s\n",
+			  ldb_dn_get_linearized(key_dn));
+		TALLOC_FREE(tmp_ctx);
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+	ldb_key = ldb_kv_key_dn(tmp_ctx, key_dn);
+	talloc_free(key_dn);
+	if (ldb_key.data == NULL) {
+		TALLOC_FREE(tmp_ctx);
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	key_dn = ldb_kv_index_key(ldb, tmp_ctx,
+				  ldb_kv, tree->u.comparison.attr,
+				  NULL, NULL, &truncation);
+	if (!key_dn) {
+		TALLOC_FREE(tmp_ctx);
+		return LDB_ERR_OPERATIONS_ERROR;
+	} else if (truncation == KEY_TRUNCATED) {
+		ldb_debug(ldb, LDB_DEBUG_WARNING,
+			  __location__
+			  ": ordered index violation: key dn truncated: %s\n",
+			  ldb_dn_get_linearized(key_dn));
+		TALLOC_FREE(tmp_ctx);
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	ldb_key2 = ldb_kv_key_dn(tmp_ctx, key_dn);
+	talloc_free(key_dn);
+	if (ldb_key2.data == NULL) {
+		TALLOC_FREE(tmp_ctx);
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	/*
+	 * In order to avoid defining a start and end key for the search, we
+	 * notice that each index key is of the form:
+	 *
+	 *     DN=@INDEX:<ATTRIBUTE>:<VALUE>\0.
+	 *
+	 * We can simply make our start key DN=@INDEX:<ATTRIBUTE>: and our end
+	 * key DN=@INDEX:<ATTRIBUTE>; to return all index entries for a
+	 * particular attribute.
+	 *
+	 * Our LMDB backend uses the default memcmp for key comparison.
+	 */
+
+	/* Eliminate NUL byte at the end of the empty key */
+	ldb_key2.length--;
+
+	if (ascending) {
+		/* : becomes ; for pseudo end-key */
+		ldb_key2.data[ldb_key2.length-1]++;
+		start_key = ldb_key;
+		end_key = ldb_key2;
+	} else {
+		start_key = ldb_key2;
+		end_key = ldb_key;
+	}
+
+	ctx.module = module;
+	ctx.error = 0;
+	ctx.dn_list = list;
+	ctx.dn_list->count = 0;
+	ctx.dn_list->dn = talloc_zero_array(ctx.dn_list, struct ldb_val, 2);
+
+	ret = ldb_kv->kv_ops->iterate_range(ldb_kv, start_key, end_key,
+					    traverse_range_index, &ctx);
+
+	if (ret != LDB_SUCCESS || ctx.error != LDB_SUCCESS) {
+		TALLOC_FREE(tmp_ctx);
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	TYPESAFE_QSORT(ctx.dn_list->dn, ctx.dn_list->count,
+		       ldb_val_equal_exact_for_qsort);
+
+	TALLOC_FREE(tmp_ctx);
+
+	return LDB_SUCCESS;
+}
+
+static int ldb_kv_index_dn_greater(struct ldb_module *module,
+				   struct ldb_kv_private *ldb_kv,
+				   const struct ldb_parse_tree *tree,
+				   struct dn_list *list)
+{
+	return ldb_kv_index_dn_ordered(module,
+				       ldb_kv,
+				       tree,
+				       list, true);
+}
+
+static int ldb_kv_index_dn_less(struct ldb_module *module,
+				   struct ldb_kv_private *ldb_kv,
+				   const struct ldb_parse_tree *tree,
+				   struct dn_list *list)
+{
+	return ldb_kv_index_dn_ordered(module,
+				       ldb_kv,
+				       tree,
+				       list, false);
+}
+
+/*
+  return a list of matching objects using a one-level index
+ */
+static int ldb_kv_index_dn_attr(struct ldb_module *module,
+				struct ldb_kv_private *ldb_kv,
+				const char *attr,
+				struct ldb_dn *dn,
+				struct dn_list *list,
+				enum key_truncation *truncation)
+{
+	struct ldb_context *ldb;
+	struct ldb_dn *key;
+	struct ldb_val val;
+	int ret;
+
+	ldb = ldb_module_get_ctx(module);
+
+	/* work out the index key from the parent DN */
+	val.data = (uint8_t *)((uintptr_t)ldb_dn_get_casefold(dn));
+	if (val.data == NULL) {
+		const char *dn_str = ldb_dn_get_linearized(dn);
+		ldb_asprintf_errstring(ldb_module_get_ctx(module),
+				       __location__
+				       ": Failed to get casefold DN "
+				       "from: %s",
+				       dn_str);
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+	val.length = strlen((char *)val.data);
+
+	/*
+	 * We use list as a TALLOC_CTX to provide a shorter-lived
+	 * memory context than ldb, even as the result is freed with
+	 * the talloc_free(key) below.
+	 */
+	key = ldb_kv_index_key(ldb, list, ldb_kv, attr, &val, NULL, truncation);
+	if (!key) {
+		ldb_oom(ldb);
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	ret = ldb_kv_dn_list_load(module, ldb_kv, key, list,
+				  DN_LIST_WILL_BE_READ_ONLY);
+	talloc_free(key);
+	if (ret != LDB_SUCCESS) {
+		return ret;
+	}
+
+	if (list->count == 0) {
+		return LDB_ERR_NO_SUCH_OBJECT;
+	}
+
+	return LDB_SUCCESS;
+}
+
+/*
+  return a list of matching objects using a one-level index
+ */
+static int ldb_kv_index_dn_one(struct ldb_module *module,
+			       struct ldb_kv_private *ldb_kv,
+			       struct ldb_dn *parent_dn,
+			       struct dn_list *list,
+			       enum key_truncation *truncation)
+{
+	int ret = ldb_kv_index_dn_attr(
+	    module, ldb_kv, LDB_KV_IDXONE, parent_dn, list, truncation);
+	if (ret == LDB_SUCCESS) {
+		/*
+		 * Ensure we do not shortcut on intersection for this
+		 * list.  We must never be lazy and return an entry
+		 * not in this list.  This allows the index for
+		 * SCOPE_ONELEVEL to be trusted.
+		 */
+
+		list->strict = true;
+	}
+	return ret;
+}
+
+/*
+  return a list of matching objects using the DN index
+ */
+static int ldb_kv_index_dn_base_dn(struct ldb_module *module,
+				   struct ldb_kv_private *ldb_kv,
+				   struct ldb_dn *base_dn,
+				   struct dn_list *dn_list,
+				   enum key_truncation *truncation)
+{
+	const struct ldb_val *guid_val = NULL;
+	if (ldb_kv->cache->GUID_index_attribute == NULL) {
+		dn_list->dn = talloc_array(dn_list, struct ldb_val, 1);
+		if (dn_list->dn == NULL) {
+			return ldb_module_oom(module);
+		}
+		dn_list->dn[0].data = discard_const_p(unsigned char,
+						      ldb_dn_get_linearized(base_dn));
+		if (dn_list->dn[0].data == NULL) {
+			talloc_free(dn_list->dn);
+			return ldb_module_oom(module);
+		}
+		dn_list->dn[0].length = strlen((char *)dn_list->dn[0].data);
+		dn_list->count = 1;
+
+		return LDB_SUCCESS;
+	}
+
+	if (ldb_kv->cache->GUID_index_dn_component != NULL) {
+		guid_val = ldb_dn_get_extended_component(
+		    base_dn, ldb_kv->cache->GUID_index_dn_component);
+	}
+
+	if (guid_val != NULL) {
+		dn_list->dn = talloc_array(dn_list, struct ldb_val, 1);
+		if (dn_list->dn == NULL) {
+			return ldb_module_oom(module);
+		}
+		dn_list->dn[0].data = guid_val->data;
+		dn_list->dn[0].length = guid_val->length;
+		dn_list->count = 1;
+
+		return LDB_SUCCESS;
+	}
+
+	return ldb_kv_index_dn_attr(
+	    module, ldb_kv, LDB_KV_IDXDN, base_dn, dn_list, truncation);
+}
+
+/*
+  return a list of dn's that might match a indexed search or
+  an error. return LDB_ERR_NO_SUCH_OBJECT for no matches, or LDB_SUCCESS for matches
+ */
+static int ldb_kv_index_dn(struct ldb_module *module,
+			   struct ldb_kv_private *ldb_kv,
+			   const struct ldb_parse_tree *tree,
+			   struct dn_list *list)
+{
+	int ret = LDB_ERR_OPERATIONS_ERROR;
+
+	switch (tree->operation) {
+	case LDB_OP_AND:
+		ret = ldb_kv_index_dn_and(module, ldb_kv, tree, list);
+		break;
+
+	case LDB_OP_OR:
+		ret = ldb_kv_index_dn_or(module, ldb_kv, tree, list);
+		break;
+
+	case LDB_OP_NOT:
+		ret = ldb_kv_index_dn_not(module, ldb_kv, tree, list);
+		break;
+
+	case LDB_OP_EQUALITY:
+		ret = ldb_kv_index_dn_leaf(module, ldb_kv, tree, list);
+		break;
+
+	case LDB_OP_GREATER:
+		ret = ldb_kv_index_dn_greater(module, ldb_kv, tree, list);
+		break;
+
+	case LDB_OP_LESS:
+		ret = ldb_kv_index_dn_less(module, ldb_kv, tree, list);
+		break;
+
+	case LDB_OP_SUBSTRING:
+	case LDB_OP_PRESENT:
+	case LDB_OP_APPROX:
+	case LDB_OP_EXTENDED:
+		/* we can't index with fancy bitops yet */
+		ret = LDB_ERR_OPERATIONS_ERROR;
+		break;
+	}
+
+	return ret;
+}
+
+/*
+  filter a candidate dn_list from an indexed search into a set of results
+  extracting just the given attributes
+*/
+static int ldb_kv_index_filter(struct ldb_kv_private *ldb_kv,
+			       const struct dn_list *dn_list,
+			       struct ldb_kv_context *ac,
+			       uint32_t *match_count,
+			       enum key_truncation scope_one_truncation)
+{
+	struct ldb_context *ldb = ldb_module_get_ctx(ac->module);
+	struct ldb_message *msg;
+	unsigned int i;
+	unsigned int num_keys = 0;
+	uint8_t previous_guid_key[LDB_KV_GUID_KEY_SIZE] = {0};
+	struct ldb_val *keys = NULL;
+
+	/*
+	 * We have to allocate the key list (rather than just walk the
+	 * caller supplied list) as the callback could change the list
+	 * (by modifying an indexed attribute hosted in the in-memory
+	 * index cache!)
+	 */
+	keys = talloc_array(ac, struct ldb_val, dn_list->count);
+	if (keys == NULL) {
+		return ldb_module_oom(ac->module);
+	}
+
+	if (ldb_kv->cache->GUID_index_attribute != NULL) {
+		/*
+		 * We speculate that the keys will be GUID based and so
+		 * pre-fill in enough space for a GUID (avoiding a pile of
+		 * small allocations)
+		 */
+		struct guid_tdb_key {
+			uint8_t guid_key[LDB_KV_GUID_KEY_SIZE];
+		} *key_values = NULL;
+
+		key_values = talloc_array(keys,
+					  struct guid_tdb_key,
+					  dn_list->count);
+
+		if (key_values == NULL) {
+			talloc_free(keys);
+			return ldb_module_oom(ac->module);
+		}
+		for (i = 0; i < dn_list->count; i++) {
+			keys[i].data = key_values[i].guid_key;
+			keys[i].length = sizeof(key_values[i].guid_key);
+		}
+	} else {
+		for (i = 0; i < dn_list->count; i++) {
+			keys[i].data = NULL;
+			keys[i].length = 0;
+		}
+	}
+
+	for (i = 0; i < dn_list->count; i++) {
+		int ret;
+
+		ret = ldb_kv_idx_to_key(
+		    ac->module, ldb_kv, keys, &dn_list->dn[i], &keys[num_keys]);
+		if (ret != LDB_SUCCESS) {
+			talloc_free(keys);
+			return ret;
+		}
+
+		if (ldb_kv->cache->GUID_index_attribute != NULL) {
+			/*
+			 * If we are in GUID index mode, then the dn_list is
+			 * sorted.  If we got a duplicate, forget about it, as
+			 * otherwise we would send the same entry back more
+			 * than once.
+			 *
+			 * This is needed in the truncated DN case, or if a
+			 * duplicate was forced in via
+			 * LDB_FLAG_INTERNAL_DISABLE_SINGLE_VALUE_CHECK
+			 */
+
+			if (memcmp(previous_guid_key,
+				   keys[num_keys].data,
+				   sizeof(previous_guid_key)) == 0) {
+				continue;
+			}
+
+			memcpy(previous_guid_key,
+			       keys[num_keys].data,
+			       sizeof(previous_guid_key));
+		}
+		num_keys++;
+	}
+
+
+	/*
+	 * Now that the list is a safe copy, send the callbacks
+	 */
+	for (i = 0; i < num_keys; i++) {
+		int ret;
+		bool matched;
+
+		/*
+		 * Check the time every 64 records, to reduce calls to
+		 * gettimeofday().  This is a compromise, not all
+		 * calls to ldb_match_message() will take the same
+		 * time, most will run quickly but by luck it might be
+		 * possible to have 64 records that are slow, doing a
+		 * recursive search via LDAP_MATCHING_RULE_IN_CHAIN.
+		 *
+		 * Thankfully this is after index processing so only
+		 * on the subset that matches some index (but still
+		 * possibly a big one like objectclass=user)
+		 */
+		if (i % 64 == 0) {
+			struct timeval now = tevent_timeval_current();
+			int timeval_cmp = tevent_timeval_compare(&ac->timeout_timeval,
+								 &now);
+
+			/*
+			 * The search has taken too long.  This is the
+			 * most likely place for our time to expire,
+			 * as we are checking the records after the
+			 * index set intersection.  This is now the
+			 * slow process of checking if the records
+			 * actually match.
+			 *
+			 * The tevent based timeout is not likely to
+			 * be hit, sadly, as we don't run an event
+			 * loop.
+			 *
+			 * While we are indexed and most of the work
+			 * should have been done already, the
+			 * ldb_match_* calls can be quite expensive if
+			 * the caller uses LDAP_MATCHING_RULE_IN_CHAIN
+			 */
+			if (timeval_cmp <= 0) {
+				talloc_free(keys);
+				return LDB_ERR_TIME_LIMIT_EXCEEDED;
+			}
+		}
+
+		msg = ldb_msg_new(ac);
+		if (!msg) {
+			talloc_free(keys);
+			return LDB_ERR_OPERATIONS_ERROR;
+		}
+
+		ret =
+		    ldb_kv_search_key(ac->module,
+				      ldb_kv,
+				      keys[i],
+				      msg,
+				      LDB_UNPACK_DATA_FLAG_NO_VALUES_ALLOC |
+				      /*
+				       * The entry point ldb_kv_search_indexed is
+				       * only called from the read-locked
+				       * ldb_kv_search.
+				       */
+				      LDB_UNPACK_DATA_FLAG_READ_LOCKED);
+		if (ret == LDB_ERR_NO_SUCH_OBJECT) {
+			/*
+			 * the record has disappeared? yes, this can
+			 * happen if the entry is deleted by something
+			 * operating in the callback (not another
+			 * process, as we have a read lock)
+			 */
+			talloc_free(msg);
+			continue;
+		}
+
+		if (ret != LDB_SUCCESS && ret != LDB_ERR_NO_SUCH_OBJECT) {
+			/* an internal error */
+			talloc_free(keys);
+			talloc_free(msg);
+			return LDB_ERR_OPERATIONS_ERROR;
+		}
+
+		/*
+		 * We trust the index for LDB_SCOPE_ONELEVEL
+		 * unless the index key has been truncated.
+		 *
+		 * LDB_SCOPE_BASE is not passed in by our only caller.
+		 */
+		if (ac->scope != LDB_SCOPE_ONELEVEL ||
+		    !ldb_kv->cache->one_level_indexes ||
+		    scope_one_truncation != KEY_NOT_TRUNCATED)
+		{
+			/*
+			 * The redaction callback may be expensive to call if it
+			 * fetches a security descriptor. Check the DN early and
+			 * bail out if it doesn't match the base.
+			 */
+			if (!ldb_match_scope(ldb, ac->base, msg->dn, ac->scope)) {
+				talloc_free(msg);
+				continue;
+			}
+		}
+
+		if (ldb->redact.callback != NULL) {
+			ret = ldb->redact.callback(ldb->redact.module, ac->req, msg);
+			if (ret != LDB_SUCCESS) {
+				talloc_free(msg);
+				return ret;
+			}
+		}
+
+		ret = ldb_match_message(ldb, msg, ac->tree,
+					ac->scope, &matched);
+		if (ret != LDB_SUCCESS) {
+			talloc_free(keys);
+			talloc_free(msg);
+			return ret;
+		}
+		if (!matched) {
+			talloc_free(msg);
+			continue;
+		}
+
+		ret = ldb_msg_add_distinguished_name(msg);
+		if (ret == -1) {
+			talloc_free(msg);
+			return LDB_ERR_OPERATIONS_ERROR;
+		}
+
+		/* filter the attributes that the user wants */
+		ret = ldb_kv_filter_attrs_in_place(msg, ac->attrs);
+		if (ret != LDB_SUCCESS) {
+			talloc_free(keys);
+			talloc_free(msg);
+			return LDB_ERR_OPERATIONS_ERROR;
+		}
+
+		ldb_msg_shrink_to_fit(msg);
+
+		/* Ensure the message elements are all talloc'd. */
+		ret = ldb_msg_elements_take_ownership(msg);
+		if (ret != LDB_SUCCESS) {
+			talloc_free(keys);
+			talloc_free(msg);
+			return LDB_ERR_OPERATIONS_ERROR;
+		}
+
+		ret = ldb_module_send_entry(ac->req, msg, NULL);
+		if (ret != LDB_SUCCESS) {
+			/* Regardless of success or failure, the msg
+			 * is the callbacks responsibility, and should
+			 * not be talloc_free()'ed */
+			ac->request_terminated = true;
+			talloc_free(keys);
+			return ret;
+		}
+
+		(*match_count)++;
+	}
+
+	TALLOC_FREE(keys);
+	return LDB_SUCCESS;
+}
+
+/*
+  sort a DN list
+ */
+static void ldb_kv_dn_list_sort(struct ldb_kv_private *ltdb,
+				struct dn_list *list)
+{
+	if (list->count < 2) {
+		return;
+	}
+
+	/* We know the list is sorted when using the GUID index */
+	if (ltdb->cache->GUID_index_attribute != NULL) {
+		return;
+	}
+
+	TYPESAFE_QSORT(list->dn, list->count,
+		       ldb_val_equal_exact_for_qsort);
+}
+
+/*
+  search the database with a LDAP-like expression using indexes
+  returns -1 if an indexed search is not possible, in which
+  case the caller should call ltdb_search_full()
+*/
+int ldb_kv_search_indexed(struct ldb_kv_context *ac, uint32_t *match_count)
+{
+	struct ldb_context *ldb = ldb_module_get_ctx(ac->module);
+	struct ldb_kv_private *ldb_kv = talloc_get_type(
+	    ldb_module_get_private(ac->module), struct ldb_kv_private);
+	struct dn_list *dn_list;
+	int ret;
+	enum ldb_scope index_scope;
+	enum key_truncation scope_one_truncation = KEY_NOT_TRUNCATED;
+
+	/* see if indexing is enabled */
+	if (!ldb_kv->cache->attribute_indexes &&
+	    !ldb_kv->cache->one_level_indexes && ac->scope != LDB_SCOPE_BASE) {
+		/* fallback to a full search */
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	dn_list = talloc_zero(ac, struct dn_list);
+	if (dn_list == NULL) {
+		return ldb_module_oom(ac->module);
+	}
+
+	/*
+	 * For the purposes of selecting the switch arm below, if we
+	 * don't have a one-level index then treat it like a subtree
+	 * search
+	 */
+	if (ac->scope == LDB_SCOPE_ONELEVEL &&
+	    !ldb_kv->cache->one_level_indexes) {
+		index_scope = LDB_SCOPE_SUBTREE;
+	} else {
+		index_scope = ac->scope;
+	}
+
+	switch (index_scope) {
+	case LDB_SCOPE_BASE:
+		/*
+		 * The only caller will have filtered the operation out
+		 * so we should never get here
+		 */
+		return ldb_operr(ldb);
+
+	case LDB_SCOPE_ONELEVEL:
+
+		/*
+		 * First, load all the one-level child objects (regardless of
+		 * whether they match the search filter or not). The database
+		 * maintains a one-level index, so retrieving this is quick.
+		 */
+		ret = ldb_kv_index_dn_one(ac->module,
+					  ldb_kv,
+					  ac->base,
+					  dn_list,
+					  &scope_one_truncation);
+		if (ret != LDB_SUCCESS) {
+			talloc_free(dn_list);
+			return ret;
+		}
+
+		/*
+		 * If we have too many children, running ldb_kv_index_filter()
+		 * over all the child objects can be quite expensive. So next
+		 * we do a separate indexed query using the search filter.
+		 *
+		 * This should be quick, but it may return objects that are not
+		 * the direct one-level child objects we're interested in.
+		 *
+		 * We only do this in the GUID index mode, which is
+		 * O(n*log(m)) otherwise the intersection below will
+		 * be too costly at O(n*m).
+		 *
+		 * We don't set a heuristic for 'too many' but instead
+		 * do it always and rely on the index lookup being
+		 * fast enough in the small case.
+		 */
+		if (ldb_kv->cache->GUID_index_attribute != NULL) {
+			struct dn_list *indexed_search_result
+				= talloc_zero(ac, struct dn_list);
+			if (indexed_search_result == NULL) {
+				talloc_free(dn_list);
+				return ldb_module_oom(ac->module);
+			}
+
+			if (!ldb_kv->cache->attribute_indexes) {
+				talloc_free(indexed_search_result);
+				talloc_free(dn_list);
+				return LDB_ERR_OPERATIONS_ERROR;
+			}
+
+			/*
+			 * Try to do an indexed database search
+			 */
+			ret = ldb_kv_index_dn(
+			    ac->module, ldb_kv, ac->tree,
+			    indexed_search_result);
+
+			/*
+			 * We can stop if we're sure the object doesn't exist
+			 */
+			if (ret == LDB_ERR_NO_SUCH_OBJECT) {
+				talloc_free(indexed_search_result);
+				talloc_free(dn_list);
+				return LDB_ERR_NO_SUCH_OBJECT;
+			}
+
+			/*
+			 * Once we have a successful search result, we
+			 * intersect it with the one-level children (dn_list).
+			 * This should give us exactly the result we're after
+			 * (we still need to run ldb_kv_index_filter() to
+			 * handle potential index truncation cases).
+			 *
+			 * The indexed search may fail because we don't support
+			 * indexing on that type of search operation, e.g.
+			 * matching against '*'. In which case we fall through
+			 * and run ldb_kv_index_filter() over all the one-level
+			 * children (which is still better than bailing out here
+			 * and falling back to a full DB scan).
+			 */
+			if (ret == LDB_SUCCESS) {
+				if (!list_intersect(ldb_kv,
+						    dn_list,
+						    indexed_search_result)) {
+					talloc_free(indexed_search_result);
+					talloc_free(dn_list);
+					return LDB_ERR_OPERATIONS_ERROR;
+				}
+			}
+		}
+		break;
+
+	case LDB_SCOPE_SUBTREE:
+	case LDB_SCOPE_DEFAULT:
+		if (!ldb_kv->cache->attribute_indexes) {
+			talloc_free(dn_list);
+			return LDB_ERR_OPERATIONS_ERROR;
+		}
+		/*
+		 * Here we load the index for the tree.  We have no
+		 * index for the subtree.
+		 */
+		ret = ldb_kv_index_dn(ac->module, ldb_kv, ac->tree, dn_list);
+		if (ret != LDB_SUCCESS) {
+			talloc_free(dn_list);
+			return ret;
+		}
+		break;
+	}
+
+	/*
+	 * It is critical that this function do the re-filter even
+	 * on things found by the index as the index can over-match
+	 * in cases of truncation (as well as when it decides it is
+	 * not worth further filtering)
+	 *
+	 * If this changes, then the index code above would need to
+	 * pass up a flag to say if any index was truncated during
+	 * processing as the truncation here refers only to the
+	 * SCOPE_ONELEVEL index.
+	 */
+	ret = ldb_kv_index_filter(
+	    ldb_kv, dn_list, ac, match_count, scope_one_truncation);
+	talloc_free(dn_list);
+	return ret;
+}
+
+/**
+ * @brief Add a DN in the index list of a given attribute name/value pair
+ *
+ * This function will add the DN in the index list for the index for
+ * the given attribute name and value.
+ *
+ * @param[in]  module       A ldb_module structure
+ *
+ * @param[in]  dn           The string representation of the DN as it
+ *                          will be stored in the index entry
+ *
+ * @param[in]  el           A ldb_message_element array, one of the entry
+ *                          referred by the v_idx is the attribute name and
+ *                          value pair which will be used to construct the
+ *                          index name
+ *
+ * @param[in]  v_idx        The index of element in the el array to use
+ *
+ * @return                  An ldb error code
+ */
+static int ldb_kv_index_add1(struct ldb_module *module,
+			     struct ldb_kv_private *ldb_kv,
+			     const struct ldb_message *msg,
+			     struct ldb_message_element *el,
+			     int v_idx)
+{
+	struct ldb_context *ldb;
+	struct ldb_dn *dn_key;
+	int ret;
+	const struct ldb_schema_attribute *a;
+	struct dn_list *list;
+	unsigned alloc_len;
+	enum key_truncation truncation = KEY_TRUNCATED;
+
+
+	ldb = ldb_module_get_ctx(module);
+
+	list = talloc_zero(module, struct dn_list);
+	if (list == NULL) {
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	dn_key = ldb_kv_index_key(ldb,
+				  list,
+				  ldb_kv,
+				  el->name,
+				  &el->values[v_idx],
+				  &a,
+				  &truncation);
+	if (!dn_key) {
+		talloc_free(list);
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+	/*
+	 * Samba only maintains unique indexes on the objectSID and objectGUID
+	 * so if a unique index key exceeds the maximum length there is a
+	 * problem.
+	 */
+	if ((truncation == KEY_TRUNCATED) && (a != NULL &&
+		(a->flags & LDB_ATTR_FLAG_UNIQUE_INDEX ||
+		(el->flags & LDB_FLAG_INTERNAL_FORCE_UNIQUE_INDEX)))) {
+
+		ldb_asprintf_errstring(
+		    ldb,
+		    __location__ ": unique index key on %s in %s, "
+				 "exceeds maximum key length of %u (encoded).",
+		    el->name,
+		    ldb_dn_get_linearized(msg->dn),
+		    ldb_kv->max_key_length);
+		talloc_free(list);
+		return LDB_ERR_CONSTRAINT_VIOLATION;
+	}
+
+	ret = ldb_kv_dn_list_load(module, ldb_kv, dn_key, list,
+				  DN_LIST_MUTABLE);
+	if (ret != LDB_SUCCESS && ret != LDB_ERR_NO_SUCH_OBJECT) {
+		talloc_free(list);
+		return ret;
+	}
+
+	/*
+	 * Check for duplicates in the @IDXDN DN -> GUID record
+	 *
+	 * This is very normal, it just means a duplicate DN creation
+	 * was attempted, so don't set the error string or print scary
+	 * messages.
+	 */
+	if (list->count > 0 &&
+	    ldb_attr_cmp(el->name, LDB_KV_IDXDN) == 0 &&
+	    truncation == KEY_NOT_TRUNCATED) {
+
+		talloc_free(list);
+		return LDB_ERR_CONSTRAINT_VIOLATION;
+
+	} else if (list->count > 0
+		   && ldb_attr_cmp(el->name, LDB_KV_IDXDN) == 0) {
+
+		/*
+		 * At least one existing entry in the DN->GUID index, which
+		 * arises when the DN indexes have been truncated
+		 *
+		 * So need to pull the DN's to check if it's really a duplicate
+		 */
+		unsigned int i;
+		for (i=0; i < list->count; i++) {
+			uint8_t guid_key[LDB_KV_GUID_KEY_SIZE];
+			struct ldb_val key = {
+				.data = guid_key,
+				.length = sizeof(guid_key)
+			};
+			const int flags = LDB_UNPACK_DATA_FLAG_NO_ATTRS;
+			struct ldb_message *rec = ldb_msg_new(ldb);
+			if (rec == NULL) {
+				return LDB_ERR_OPERATIONS_ERROR;
+			}
+
+			ret = ldb_kv_idx_to_key(
+			    module, ldb_kv, ldb, &list->dn[i], &key);
+			if (ret != LDB_SUCCESS) {
+				TALLOC_FREE(list);
+				TALLOC_FREE(rec);
+				return ret;
+			}
+
+			ret =
+			    ldb_kv_search_key(module, ldb_kv, key, rec, flags);
+			if (key.data != guid_key) {
+				TALLOC_FREE(key.data);
+			}
+			if (ret == LDB_ERR_NO_SUCH_OBJECT) {
+				/*
+				 * the record has disappeared?
+				 * yes, this can happen
+				 */
+				talloc_free(rec);
+				continue;
+			}
+
+			if (ret != LDB_SUCCESS) {
+				/* an internal error */
+				TALLOC_FREE(rec);
+				TALLOC_FREE(list);
+				return LDB_ERR_OPERATIONS_ERROR;
+			}
+			/*
+			 * The DN we are trying to add to the DB and index
+			 * is already here, so we must deny the addition
+			 */
+			if (ldb_dn_compare(msg->dn, rec->dn) == 0) {
+				TALLOC_FREE(rec);
+				TALLOC_FREE(list);
+				return LDB_ERR_CONSTRAINT_VIOLATION;
+			}
+		}
+	}
+
+	/*
+	 * Check for duplicates in unique indexes
+	 *
+	 * We don't need to do a loop test like the @IDXDN case
+	 * above as we have a ban on long unique index values
+	 * at the start of this function.
+	 */
+	if (list->count > 0 &&
+	    ((a != NULL
+	      && (a->flags & LDB_ATTR_FLAG_UNIQUE_INDEX ||
+		  (el->flags & LDB_FLAG_INTERNAL_FORCE_UNIQUE_INDEX))))) {
+		/*
+		 * We do not want to print info about a possibly
+		 * confidential DN that the conflict was with in the
+		 * user-visible error string
+		 */
+
+		if (ldb_kv->cache->GUID_index_attribute == NULL) {
+			ldb_debug(ldb, LDB_DEBUG_WARNING,
+				  __location__
+				  ": unique index violation on %s in %s, "
+				  "conflicts with %*.*s in %s",
+				  el->name, ldb_dn_get_linearized(msg->dn),
+				  (int)list->dn[0].length,
+				  (int)list->dn[0].length,
+				  list->dn[0].data,
+				  ldb_dn_get_linearized(dn_key));
+		} else {
+			/* This can't fail, gives a default at worst */
+			const struct ldb_schema_attribute *attr =
+			    ldb_schema_attribute_by_name(
+				ldb, ldb_kv->cache->GUID_index_attribute);
+			struct ldb_val v;
+			ret = attr->syntax->ldif_write_fn(ldb, list,
+							  &list->dn[0], &v);
+			if (ret == LDB_SUCCESS) {
+				ldb_debug(ldb,
+					  LDB_DEBUG_WARNING,
+					  __location__
+					  ": unique index violation on %s in "
+					  "%s, conflicts with %s %*.*s in %s",
+					  el->name,
+					  ldb_dn_get_linearized(msg->dn),
+					  ldb_kv->cache->GUID_index_attribute,
+					  (int)v.length,
+					  (int)v.length,
+					  v.data,
+					  ldb_dn_get_linearized(dn_key));
+			}
+		}
+		ldb_asprintf_errstring(ldb,
+				       __location__ ": unique index violation "
+				       "on %s in %s",
+				       el->name,
+				       ldb_dn_get_linearized(msg->dn));
+		talloc_free(list);
+		return LDB_ERR_CONSTRAINT_VIOLATION;
+	}
+
+	/* overallocate the list a bit, to reduce the number of
+	 * realloc triggered copies */
+	alloc_len = ((list->count+1)+7) & ~7;
+	list->dn = talloc_realloc(list, list->dn, struct ldb_val, alloc_len);
+	if (list->dn == NULL) {
+		talloc_free(list);
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	if (ldb_kv->cache->GUID_index_attribute == NULL) {
+		const char *dn_str = ldb_dn_get_linearized(msg->dn);
+		list->dn[list->count].data
+			= (uint8_t *)talloc_strdup(list->dn, dn_str);
+		if (list->dn[list->count].data == NULL) {
+			talloc_free(list);
+			return LDB_ERR_OPERATIONS_ERROR;
+		}
+		list->dn[list->count].length = strlen(dn_str);
+	} else {
+		const struct ldb_val *key_val;
+		struct ldb_val *exact = NULL, *next = NULL;
+		key_val = ldb_msg_find_ldb_val(
+		    msg, ldb_kv->cache->GUID_index_attribute);
+		if (key_val == NULL) {
+			talloc_free(list);
+			return ldb_module_operr(module);
+		}
+
+		if (key_val->length != LDB_KV_GUID_SIZE) {
+			talloc_free(list);
+			return ldb_module_operr(module);
+		}
+
+		BINARY_ARRAY_SEARCH_GTE(list->dn, list->count,
+					*key_val, ldb_val_equal_exact_ordered,
+					exact, next);
+
+		/*
+		 * Give a warning rather than fail, this could be a
+		 * duplicate value in the record allowed by a caller
+		 * forcing in the value with
+		 * LDB_FLAG_INTERNAL_DISABLE_SINGLE_VALUE_CHECK
+		 */
+		if (exact != NULL && truncation == KEY_NOT_TRUNCATED) {
+			/* This can't fail, gives a default at worst */
+			const struct ldb_schema_attribute *attr =
+			    ldb_schema_attribute_by_name(
+				ldb, ldb_kv->cache->GUID_index_attribute);
+			struct ldb_val v;
+			ret = attr->syntax->ldif_write_fn(ldb, list,
+							  exact, &v);
+			if (ret == LDB_SUCCESS) {
+				ldb_debug(ldb,
+					  LDB_DEBUG_WARNING,
+					  __location__
+					  ": duplicate attribute value in %s "
+					  "for index on %s, "
+					  "duplicate of %s %*.*s in %s",
+					  ldb_dn_get_linearized(msg->dn),
+					  el->name,
+					  ldb_kv->cache->GUID_index_attribute,
+					  (int)v.length,
+					  (int)v.length,
+					  v.data,
+					  ldb_dn_get_linearized(dn_key));
+			}
+		}
+
+		if (next == NULL) {
+			next = &list->dn[list->count];
+		} else {
+			memmove(&next[1], next,
+				sizeof(*next) * (list->count - (next - list->dn)));
+		}
+		*next = ldb_val_dup(list->dn, key_val);
+		if (next->data == NULL) {
+			talloc_free(list);
+			return ldb_module_operr(module);
+		}
+	}
+	list->count++;
+
+	ret = ldb_kv_dn_list_store(module, dn_key, list);
+
+	talloc_free(list);
+
+	return ret;
+}
+
+/*
+  add index entries for one elements in a message
+ */
+static int ldb_kv_index_add_el(struct ldb_module *module,
+			       struct ldb_kv_private *ldb_kv,
+			       const struct ldb_message *msg,
+			       struct ldb_message_element *el)
+{
+	unsigned int i;
+	for (i = 0; i < el->num_values; i++) {
+		int ret = ldb_kv_index_add1(module, ldb_kv, msg, el, i);
+		if (ret != LDB_SUCCESS) {
+			return ret;
+		}
+	}
+
+	return LDB_SUCCESS;
+}
+
+/*
+  add index entries for all elements in a message
+ */
+static int ldb_kv_index_add_all(struct ldb_module *module,
+				struct ldb_kv_private *ldb_kv,
+				const struct ldb_message *msg)
+{
+	struct ldb_message_element *elements = msg->elements;
+	unsigned int i;
+	const char *dn_str;
+	int ret;
+
+	if (ldb_dn_is_special(msg->dn)) {
+		return LDB_SUCCESS;
+	}
+
+	dn_str = ldb_dn_get_linearized(msg->dn);
+	if (dn_str == NULL) {
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	ret = ldb_kv_write_index_dn_guid(module, msg, 1);
+	if (ret != LDB_SUCCESS) {
+		return ret;
+	}
+
+	if (!ldb_kv->cache->attribute_indexes) {
+		/* no indexed fields */
+		return LDB_SUCCESS;
+	}
+
+	for (i = 0; i < msg->num_elements; i++) {
+		if (!ldb_kv_is_indexed(module, ldb_kv, elements[i].name)) {
+			continue;
+		}
+		ret = ldb_kv_index_add_el(module, ldb_kv, msg, &elements[i]);
+		if (ret != LDB_SUCCESS) {
+			struct ldb_context *ldb = ldb_module_get_ctx(module);
+			ldb_asprintf_errstring(ldb,
+					       __location__ ": Failed to re-index %s in %s - %s",
+					       elements[i].name, dn_str,
+					       ldb_errstring(ldb));
+			return ret;
+		}
+	}
+
+	return LDB_SUCCESS;
+}
+
+
+/*
+  insert a DN index for a message
+*/
+static int ldb_kv_modify_index_dn(struct ldb_module *module,
+				  struct ldb_kv_private *ldb_kv,
+				  const struct ldb_message *msg,
+				  struct ldb_dn *dn,
+				  const char *index,
+				  int add)
+{
+	struct ldb_message_element el;
+	struct ldb_val val;
+	int ret;
+
+	val.data = (uint8_t *)((uintptr_t)ldb_dn_get_casefold(dn));
+	if (val.data == NULL) {
+		const char *dn_str = ldb_dn_get_linearized(dn);
+		ldb_asprintf_errstring(ldb_module_get_ctx(module),
+				       __location__ ": Failed to modify %s "
+						    "against %s in %s: failed "
+						    "to get casefold DN",
+				       index,
+				       ldb_kv->cache->GUID_index_attribute,
+				       dn_str);
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	val.length = strlen((char *)val.data);
+	el.name = index;
+	el.values = &val;
+	el.num_values = 1;
+
+	if (add) {
+		ret = ldb_kv_index_add1(module, ldb_kv, msg, &el, 0);
+	} else { /* delete */
+		ret = ldb_kv_index_del_value(module, ldb_kv, msg, &el, 0);
+	}
+
+	if (ret != LDB_SUCCESS) {
+		struct ldb_context *ldb = ldb_module_get_ctx(module);
+		const char *dn_str = ldb_dn_get_linearized(dn);
+		ldb_asprintf_errstring(ldb,
+				       __location__ ": Failed to modify %s "
+						    "against %s in %s - %s",
+				       index,
+				       ldb_kv->cache->GUID_index_attribute,
+				       dn_str,
+				       ldb_errstring(ldb));
+		return ret;
+	}
+	return ret;
+}
+
+/*
+  insert a one level index for a message
+*/
+static int ldb_kv_index_onelevel(struct ldb_module *module,
+				 const struct ldb_message *msg,
+				 int add)
+{
+	struct ldb_kv_private *ldb_kv = talloc_get_type(
+	    ldb_module_get_private(module), struct ldb_kv_private);
+	struct ldb_dn *pdn;
+	int ret;
+
+	/* We index for ONE Level only if requested */
+	if (!ldb_kv->cache->one_level_indexes) {
+		return LDB_SUCCESS;
+	}
+
+	pdn = ldb_dn_get_parent(module, msg->dn);
+	if (pdn == NULL) {
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+	ret =
+	    ldb_kv_modify_index_dn(module, ldb_kv, msg, pdn, LDB_KV_IDXONE, add);
+
+	talloc_free(pdn);
+
+	return ret;
+}
+
+/*
+  insert a one level index for a message
+*/
+static int ldb_kv_write_index_dn_guid(struct ldb_module *module,
+				      const struct ldb_message *msg,
+				      int add)
+{
+	int ret;
+	struct ldb_kv_private *ldb_kv = talloc_get_type(
+	    ldb_module_get_private(module), struct ldb_kv_private);
+
+	/* We index for DN only if using a GUID index */
+	if (ldb_kv->cache->GUID_index_attribute == NULL) {
+		return LDB_SUCCESS;
+	}
+
+	ret = ldb_kv_modify_index_dn(
+	    module, ldb_kv, msg, msg->dn, LDB_KV_IDXDN, add);
+
+	if (ret == LDB_ERR_CONSTRAINT_VIOLATION) {
+		ldb_asprintf_errstring(ldb_module_get_ctx(module),
+				       "Entry %s already exists",
+				       ldb_dn_get_linearized(msg->dn));
+		ret = LDB_ERR_ENTRY_ALREADY_EXISTS;
+	}
+	return ret;
+}
+
+/*
+  add the index entries for a new element in a record
+  The caller guarantees that these element values are not yet indexed
+*/
+int ldb_kv_index_add_element(struct ldb_module *module,
+			     struct ldb_kv_private *ldb_kv,
+			     const struct ldb_message *msg,
+			     struct ldb_message_element *el)
+{
+	if (ldb_dn_is_special(msg->dn)) {
+		return LDB_SUCCESS;
+	}
+	if (!ldb_kv_is_indexed(module, ldb_kv, el->name)) {
+		return LDB_SUCCESS;
+	}
+	return ldb_kv_index_add_el(module, ldb_kv, msg, el);
+}
+
+/*
+  add the index entries for a new record
+*/
+int ldb_kv_index_add_new(struct ldb_module *module,
+			 struct ldb_kv_private *ldb_kv,
+			 const struct ldb_message *msg)
+{
+	int ret;
+
+	if (ldb_dn_is_special(msg->dn)) {
+		return LDB_SUCCESS;
+	}
+
+	ret = ldb_kv_index_add_all(module, ldb_kv, msg);
+	if (ret != LDB_SUCCESS) {
+		/*
+		 * Because we can't trust the caller to be doing
+		 * transactions properly, clean up any index for this
+		 * entry rather than relying on a transaction
+		 * cleanup
+		 */
+
+		ldb_kv_index_delete(module, msg);
+		return ret;
+	}
+
+	ret = ldb_kv_index_onelevel(module, msg, 1);
+	if (ret != LDB_SUCCESS) {
+		/*
+		 * Because we can't trust the caller to be doing
+		 * transactions properly, clean up any index for this
+		 * entry rather than relying on a transaction
+		 * cleanup
+		 */
+		ldb_kv_index_delete(module, msg);
+		return ret;
+	}
+	return ret;
+}
+
+
+/*
+  delete an index entry for one message element
+*/
+int ldb_kv_index_del_value(struct ldb_module *module,
+			   struct ldb_kv_private *ldb_kv,
+			   const struct ldb_message *msg,
+			   struct ldb_message_element *el,
+			   unsigned int v_idx)
+{
+	struct ldb_context *ldb;
+	struct ldb_dn *dn_key;
+	const char *dn_str;
+	int ret, i;
+	unsigned int j;
+	struct dn_list *list;
+	struct ldb_dn *dn = msg->dn;
+	enum key_truncation truncation = KEY_NOT_TRUNCATED;
+
+	ldb = ldb_module_get_ctx(module);
+
+	dn_str = ldb_dn_get_linearized(dn);
+	if (dn_str == NULL) {
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	if (dn_str[0] == '@') {
+		return LDB_SUCCESS;
+	}
+
+	/*
+	 * ldb is being used as the memory context to ldb_kv_index_key
+	 * as dn_key itself is also used as the TALLOC_CTX for the
+	 * rest of this function.
+	 */
+	dn_key = ldb_kv_index_key(ldb,
+				  ldb,
+				  ldb_kv,
+				  el->name,
+				  &el->values[v_idx],
+				  NULL,
+				  &truncation);
+	/*
+	 * We ignore key truncation in ltdb_index_add1() so
+	 * match that by ignoring it here as well
+	 *
+	 * Multiple values are legitimate and accepted
+	 */
+	if (!dn_key) {
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	list = talloc_zero(dn_key, struct dn_list);
+	if (list == NULL) {
+		talloc_free(dn_key);
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	ret = ldb_kv_dn_list_load(module, ldb_kv, dn_key, list,
+				  DN_LIST_MUTABLE);
+	if (ret == LDB_ERR_NO_SUCH_OBJECT) {
+		/* it wasn't indexed. Did we have an earlier error? If we did then
+		   its gone now */
+		talloc_free(dn_key);
+		return LDB_SUCCESS;
+	}
+
+	if (ret != LDB_SUCCESS) {
+		talloc_free(dn_key);
+		return ret;
+	}
+
+	/*
+	 * Find one of the values matching this message to remove
+	 */
+	i = ldb_kv_dn_list_find_msg(ldb_kv, list, msg);
+	if (i == -1) {
+		/* nothing to delete */
+		talloc_free(dn_key);
+		return LDB_SUCCESS;
+	}
+
+	j = (unsigned int) i;
+	ARRAY_DEL_ELEMENT(list->dn, j, list->count);
+	list->count--;
+	if (list->count == 0) {
+		talloc_free(list->dn);
+		list->dn = NULL;
+	} else {
+		list->dn = talloc_realloc(list, list->dn, struct ldb_val, list->count);
+	}
+
+	ret = ldb_kv_dn_list_store(module, dn_key, list);
+
+	talloc_free(dn_key);
+
+	return ret;
+}
+
+/*
+  delete the index entries for a element
+  return -1 on failure
+*/
+int ldb_kv_index_del_element(struct ldb_module *module,
+			     struct ldb_kv_private *ldb_kv,
+			     const struct ldb_message *msg,
+			     struct ldb_message_element *el)
+{
+	const char *dn_str;
+	int ret;
+	unsigned int i;
+
+	if (!ldb_kv->cache->attribute_indexes) {
+		/* no indexed fields */
+		return LDB_SUCCESS;
+	}
+
+	dn_str = ldb_dn_get_linearized(msg->dn);
+	if (dn_str == NULL) {
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	if (dn_str[0] == '@') {
+		return LDB_SUCCESS;
+	}
+
+	if (!ldb_kv_is_indexed(module, ldb_kv, el->name)) {
+		return LDB_SUCCESS;
+	}
+	for (i = 0; i < el->num_values; i++) {
+		ret = ldb_kv_index_del_value(module, ldb_kv, msg, el, i);
+		if (ret != LDB_SUCCESS) {
+			return ret;
+		}
+	}
+
+	return LDB_SUCCESS;
+}
+
+/*
+  delete the index entries for a record
+  return -1 on failure
+*/
+int ldb_kv_index_delete(struct ldb_module *module,
+			const struct ldb_message *msg)
+{
+	struct ldb_kv_private *ldb_kv = talloc_get_type(
+	    ldb_module_get_private(module), struct ldb_kv_private);
+	int ret;
+	unsigned int i;
+
+	if (ldb_dn_is_special(msg->dn)) {
+		return LDB_SUCCESS;
+	}
+
+	ret = ldb_kv_index_onelevel(module, msg, 0);
+	if (ret != LDB_SUCCESS) {
+		return ret;
+	}
+
+	ret = ldb_kv_write_index_dn_guid(module, msg, 0);
+	if (ret != LDB_SUCCESS) {
+		return ret;
+	}
+
+	if (!ldb_kv->cache->attribute_indexes) {
+		/* no indexed fields */
+		return LDB_SUCCESS;
+	}
+
+	for (i = 0; i < msg->num_elements; i++) {
+		ret = ldb_kv_index_del_element(
+		    module, ldb_kv, msg, &msg->elements[i]);
+		if (ret != LDB_SUCCESS) {
+			return ret;
+		}
+	}
+
+	return LDB_SUCCESS;
+}
+
+
+/*
+  traversal function that deletes all @INDEX records in the in-memory
+  TDB.
+
+  This does not touch the actual DB, that is done at transaction
+  commit, which in turn greatly reduces DB churn as we will likely
+  be able to do a direct update into the old record.
+*/
+static int delete_index(struct ldb_kv_private *ldb_kv,
+			struct ldb_val key,
+			_UNUSED_ struct ldb_val data,
+			void *state)
+{
+	struct ldb_module *module = state;
+	const char *dnstr = "DN=" LDB_KV_INDEX ":";
+	struct dn_list list;
+	struct ldb_dn *dn;
+	struct ldb_val v;
+	int ret;
+
+	if (strncmp((char *)key.data, dnstr, strlen(dnstr)) != 0) {
+		return 0;
+	}
+	/* we need to put a empty list in the internal tdb for this
+	 * index entry */
+	list.dn = NULL;
+	list.count = 0;
+
+	/* the offset of 3 is to remove the DN= prefix. */
+	v.data = key.data + 3;
+	v.length = strnlen((char *)key.data, key.length) - 3;
+
+	dn = ldb_dn_from_ldb_val(ldb_kv, ldb_module_get_ctx(module), &v);
+
+	/*
+	 * This does not actually touch the DB quite yet, just
+         * the in-memory index cache
+	 */
+	ret = ldb_kv_dn_list_store(module, dn, &list);
+	if (ret != LDB_SUCCESS) {
+		ldb_asprintf_errstring(ldb_module_get_ctx(module),
+				       "Unable to store null index for %s\n",
+						ldb_dn_get_linearized(dn));
+		talloc_free(dn);
+		return -1;
+	}
+	talloc_free(dn);
+	return 0;
+}
+
+/*
+  traversal function that adds @INDEX records during a re index TODO wrong comment
+*/
+static int re_key(struct ldb_kv_private *ldb_kv,
+		  struct ldb_val key,
+		  struct ldb_val val,
+		  void *state)
+{
+	struct ldb_context *ldb;
+	struct ldb_kv_reindex_context *ctx =
+	    (struct ldb_kv_reindex_context *)state;
+	struct ldb_module *module = ldb_kv->module;
+	struct ldb_message *msg;
+	int ret;
+	struct ldb_val key2;
+	bool is_record;
+
+	ldb = ldb_module_get_ctx(module);
+
+	is_record = ldb_kv_key_is_normal_record(key);
+	if (is_record == false) {
+		return 0;
+	}
+
+	msg = ldb_msg_new(module);
+	if (msg == NULL) {
+		return -1;
+	}
+
+	ret = ldb_unpack_data(ldb, &val, msg);
+	if (ret != 0) {
+		ldb_debug(ldb, LDB_DEBUG_ERROR, "Invalid data for index %s\n",
+						ldb_dn_get_linearized(msg->dn));
+		ctx->error = ret;
+		talloc_free(msg);
+		return -1;
+	}
+
+	if (msg->dn == NULL) {
+		ldb_debug(ldb, LDB_DEBUG_ERROR,
+			  "Refusing to re-index as GUID "
+			  "key %*.*s with no DN\n",
+			  (int)key.length, (int)key.length,
+			  (char *)key.data);
+		talloc_free(msg);
+		return -1;
+	}
+
+	/* check if the DN key has changed, perhaps due to the case
+	   insensitivity of an element changing, or a change from DN
+	   to GUID keys */
+	key2 = ldb_kv_key_msg(module, msg, msg);
+	if (key2.data == NULL) {
+		/* probably a corrupt record ... darn */
+		ldb_debug(ldb, LDB_DEBUG_ERROR, "Invalid DN in re_index: %s",
+						ldb_dn_get_linearized(msg->dn));
+		talloc_free(msg);
+		return 0;
+	}
+	if (key.length != key2.length ||
+	    (memcmp(key.data, key2.data, key.length) != 0)) {
+		ldb_kv->kv_ops->update_in_iterate(
+		    ldb_kv, key, key2, val, ctx);
+	}
+	talloc_free(key2.data);
+
+	talloc_free(msg);
+
+	ctx->count++;
+	if (ctx->count % 10000 == 0) {
+		ldb_debug(ldb, LDB_DEBUG_WARNING,
+			  "Reindexing: re-keyed %u records so far",
+			  ctx->count);
+	}
+
+	return 0;
+}
+
+/*
+  traversal function that adds @INDEX records during a re index
+*/
+static int re_index(struct ldb_kv_private *ldb_kv,
+		    struct ldb_val key,
+		    struct ldb_val val,
+		    void *state)
+{
+	struct ldb_context *ldb;
+	struct ldb_kv_reindex_context *ctx =
+	    (struct ldb_kv_reindex_context *)state;
+	struct ldb_module *module = ldb_kv->module;
+	struct ldb_message *msg;
+	int ret;
+	bool is_record;
+
+	ldb = ldb_module_get_ctx(module);
+
+	is_record = ldb_kv_key_is_normal_record(key);
+	if (is_record == false) {
+		return 0;
+	}
+
+	msg = ldb_msg_new(module);
+	if (msg == NULL) {
+		return -1;
+	}
+
+	ret = ldb_unpack_data(ldb, &val, msg);
+	if (ret != 0) {
+		ldb_debug(ldb, LDB_DEBUG_ERROR, "Invalid data for index %s\n",
+						ldb_dn_get_linearized(msg->dn));
+		ctx->error = ret;
+		talloc_free(msg);
+		return -1;
+	}
+
+	if (msg->dn == NULL) {
+		ldb_debug(ldb, LDB_DEBUG_ERROR,
+			  "Refusing to re-index as GUID "
+			  "key %*.*s with no DN\n",
+			  (int)key.length, (int)key.length,
+			  (char *)key.data);
+		talloc_free(msg);
+		return -1;
+	}
+
+	ret = ldb_kv_index_onelevel(module, msg, 1);
+	if (ret != LDB_SUCCESS) {
+		ldb_debug(ldb, LDB_DEBUG_ERROR,
+			  "Adding special ONE LEVEL index failed (%s)!",
+						ldb_dn_get_linearized(msg->dn));
+		talloc_free(msg);
+		return -1;
+	}
+
+	ret = ldb_kv_index_add_all(module, ldb_kv, msg);
+
+	if (ret != LDB_SUCCESS) {
+		ctx->error = ret;
+		talloc_free(msg);
+		return -1;
+	}
+
+	talloc_free(msg);
+
+	ctx->count++;
+	if (ctx->count % 10000 == 0) {
+		ldb_debug(ldb, LDB_DEBUG_WARNING,
+			  "Reindexing: re-indexed %u records so far",
+			  ctx->count);
+	}
+
+	return 0;
+}
+
+/*
+ * Convert the 4-byte pack format version to a number that's slightly
+ * more intelligible to a user e.g. version 0, 1, 2, etc.
+ */
+static uint32_t displayable_pack_version(uint32_t version) {
+	if (version < LDB_PACKING_FORMAT_NODN) {
+		return version; /* unknown - can't convert */
+	}
+
+	return (version - LDB_PACKING_FORMAT_NODN);
+}
+
+static int re_pack(struct ldb_kv_private *ldb_kv,
+		   _UNUSED_ struct ldb_val key,
+		   struct ldb_val val,
+		   void *state)
+{
+	struct ldb_context *ldb;
+	struct ldb_message *msg;
+	struct ldb_module *module = ldb_kv->module;
+	struct ldb_kv_repack_context *ctx =
+	    (struct ldb_kv_repack_context *)state;
+	int ret;
+
+	ldb = ldb_module_get_ctx(module);
+
+	msg = ldb_msg_new(module);
+	if (msg == NULL) {
+		return -1;
+	}
+
+	ret = ldb_unpack_data(ldb, &val, msg);
+	if (ret != 0) {
+		ldb_debug(ldb, LDB_DEBUG_ERROR, "Repack: unpack failed: %s\n",
+			  ldb_dn_get_linearized(msg->dn));
+		ctx->error = ret;
+		talloc_free(msg);
+		return -1;
+	}
+
+	ret = ldb_kv_store(module, msg, TDB_MODIFY);
+	if (ret != LDB_SUCCESS) {
+		ldb_debug(ldb, LDB_DEBUG_ERROR, "Repack: store failed: %s\n",
+			  ldb_dn_get_linearized(msg->dn));
+		ctx->error = ret;
+		talloc_free(msg);
+		return -1;
+	}
+
+	/*
+	 * Warn the user that we're repacking the first time we see a normal
+	 * record. This means we never warn if we're repacking a database with
+	 * only @ records. This is because during database initialisation,
+	 * we might repack as initial settings are written out, and we don't
+	 * want to spam the log.
+	 */
+	if ((!ctx->normal_record_seen) && (!ldb_dn_is_special(msg->dn))) {
+		ldb_debug(ldb, LDB_DEBUG_ALWAYS_LOG,
+			  "Repacking database from v%u to v%u format "
+			  "(first record %s)",
+			  displayable_pack_version(ctx->old_version),
+			  displayable_pack_version(ldb_kv->pack_format_version),
+			  ldb_dn_get_linearized(msg->dn));
+		ctx->normal_record_seen = true;
+	}
+
+	ctx->count++;
+	if (ctx->count % 10000 == 0) {
+		ldb_debug(ldb, LDB_DEBUG_WARNING,
+			  "Repack: re-packed %u records so far",
+			  ctx->count);
+	}
+
+	talloc_free(msg);
+	return 0;
+}
+
+int ldb_kv_repack(struct ldb_module *module)
+{
+	struct ldb_kv_private *ldb_kv = talloc_get_type(
+	    ldb_module_get_private(module), struct ldb_kv_private);
+	struct ldb_context *ldb = ldb_module_get_ctx(module);
+	struct ldb_kv_repack_context ctx;
+	int ret;
+
+	ctx.old_version = ldb_kv->pack_format_version;
+	ctx.count = 0;
+	ctx.error = LDB_SUCCESS;
+	ctx.normal_record_seen = false;
+
+	ldb_kv->pack_format_version = ldb_kv->target_pack_format_version;
+
+	/* Iterate all database records and repack them in the new format */
+	ret = ldb_kv->kv_ops->iterate(ldb_kv, re_pack, &ctx);
+	if (ret < 0) {
+		ldb_debug(ldb, LDB_DEBUG_ERROR, "Repack traverse failed: %s",
+			  ldb_errstring(ldb));
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	if (ctx.error != LDB_SUCCESS) {
+		ldb_debug(ldb, LDB_DEBUG_ERROR, "Repack failed: %s",
+			  ldb_errstring(ldb));
+		return ctx.error;
+	}
+
+	return LDB_SUCCESS;
+}
+
+/*
+  force a complete reindex of the database
+*/
+int ldb_kv_reindex(struct ldb_module *module)
+{
+	struct ldb_kv_private *ldb_kv = talloc_get_type(
+	    ldb_module_get_private(module), struct ldb_kv_private);
+	int ret;
+	struct ldb_kv_reindex_context ctx;
+	size_t index_cache_size = 0;
+
+	/*
+	 * Only triggered after a modification, but make clear we do
+	 * not re-index a read-only DB
+	 */
+	if (ldb_kv->read_only) {
+		return LDB_ERR_UNWILLING_TO_PERFORM;
+	}
+
+	if (ldb_kv_cache_reload(module) != 0) {
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	/*
+	 * Ensure we read (and so remove) the entries from the real
+	 * DB, no values stored so far are any use as we want to do a
+	 * re-index
+	 */
+	ldb_kv_index_transaction_cancel(module);
+	if (ldb_kv->nested_idx_ptr != NULL) {
+		ldb_kv_index_sub_transaction_cancel(ldb_kv);
+	}
+
+	/*
+	 * Calculate the size of the index cache needed for
+	 * the re-index. If specified always use the
+	 * ldb_kv->index_transaction_cache_size otherwise use the maximum
+	 * of the size estimate or the DEFAULT_INDEX_CACHE_SIZE
+	 */
+	if (ldb_kv->index_transaction_cache_size > 0) {
+		index_cache_size = ldb_kv->index_transaction_cache_size;
+	} else {
+		index_cache_size = ldb_kv->kv_ops->get_size(ldb_kv);
+		if (index_cache_size < DEFAULT_INDEX_CACHE_SIZE) {
+			index_cache_size = DEFAULT_INDEX_CACHE_SIZE;
+		}
+	}
+
+	/*
+	 * Note that we don't start an index sub transaction for re-indexing
+	 */
+	ret = ldb_kv_index_transaction_start(module, index_cache_size);
+	if (ret != LDB_SUCCESS) {
+		return ret;
+	}
+
+	/* first traverse the database deleting any @INDEX records by
+	 * putting NULL entries in the in-memory tdb
+	 */
+	ret = ldb_kv->kv_ops->iterate(ldb_kv, delete_index, module);
+	if (ret < 0) {
+		struct ldb_context *ldb = ldb_module_get_ctx(module);
+		ldb_asprintf_errstring(ldb, "index deletion traverse failed: %s",
+				       ldb_errstring(ldb));
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	ctx.error = 0;
+	ctx.count = 0;
+
+	ret = ldb_kv->kv_ops->iterate(ldb_kv, re_key, &ctx);
+	if (ret < 0) {
+		struct ldb_context *ldb = ldb_module_get_ctx(module);
+		ldb_asprintf_errstring(ldb, "key correction traverse failed: %s",
+				       ldb_errstring(ldb));
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	if (ctx.error != LDB_SUCCESS) {
+		struct ldb_context *ldb = ldb_module_get_ctx(module);
+		ldb_asprintf_errstring(ldb, "reindexing failed: %s", ldb_errstring(ldb));
+		return ctx.error;
+	}
+
+	ctx.error = 0;
+	ctx.count = 0;
+
+	/* now traverse adding any indexes for normal LDB records */
+	ret = ldb_kv->kv_ops->iterate(ldb_kv, re_index, &ctx);
+	if (ret < 0) {
+		struct ldb_context *ldb = ldb_module_get_ctx(module);
+		ldb_asprintf_errstring(ldb, "reindexing traverse failed: %s",
+				       ldb_errstring(ldb));
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	if (ctx.error != LDB_SUCCESS) {
+		struct ldb_context *ldb = ldb_module_get_ctx(module);
+		ldb_asprintf_errstring(ldb, "reindexing failed: %s", ldb_errstring(ldb));
+		return ctx.error;
+	}
+
+	if (ctx.count > 10000) {
+		ldb_debug(ldb_module_get_ctx(module),
+			  LDB_DEBUG_WARNING,
+			  "Reindexing: re_index successful on %s, "
+			  "final index write-out will be in transaction commit",
+			  ldb_kv->kv_ops->name(ldb_kv));
+	}
+	return LDB_SUCCESS;
+}
+
+/*
+ * Copy the contents of the nested transaction index cache record to the
+ * transaction index cache.
+ *
+ * During this 'commit' of the subtransaction to the main transaction
+ * (cache), care must be taken to free any existing index at the top
+ * level because otherwise we would leak memory.
+ */
+static int ldb_kv_sub_transaction_traverse(
+	struct tdb_context *tdb,
+	TDB_DATA key,
+	TDB_DATA data,
+	void *state)
+{
+	struct ldb_module *module = state;
+	struct ldb_kv_private *ldb_kv = talloc_get_type(
+	    ldb_module_get_private(module), struct ldb_kv_private);
+	TDB_DATA rec = {0};
+	struct dn_list *index_in_subtransaction = NULL;
+	struct dn_list *index_in_top_level = NULL;
+	int ret = 0;
+
+	/*
+	 * This unwraps the pointer in the DB into a pointer in
+	 * memory, we are abusing TDB as a hash map, not a linearised
+	 * database store
+	 */
+	index_in_subtransaction = ldb_kv_index_idxptr(module, data);
+	if (index_in_subtransaction == NULL) {
+		ldb_kv->idxptr->error = LDB_ERR_OPERATIONS_ERROR;
+		return -1;
+	}
+
+	/*
+	 * Do we already have an entry in the primary transaction cache
+	 * If so free it's dn_list and replace it with the dn_list from
+	 * the secondary cache
+	 *
+	 * The TDB and so the fetched rec contains NO DATA, just a
+	 * pointer to data held in memory.
+	 */
+	rec = tdb_fetch(ldb_kv->idxptr->itdb, key);
+	if (rec.dptr != NULL) {
+		index_in_top_level = ldb_kv_index_idxptr(module, rec);
+		free(rec.dptr);
+		if (index_in_top_level == NULL) {
+			abort();
+		}
+		/*
+		 * We had this key at the top level.  However we made a copy
+		 * at the sub-transaction level so that we could possibly
+		 * roll back.  We have to free the top level index memory
+		 * otherwise we would leak
+		 */
+		if (index_in_top_level->count > 0) {
+			TALLOC_FREE(index_in_top_level->dn);
+		}
+		index_in_top_level->dn
+			= talloc_steal(index_in_top_level,
+				       index_in_subtransaction->dn);
+		index_in_top_level->count = index_in_subtransaction->count;
+		return 0;
+	}
+
+	index_in_top_level = talloc(ldb_kv->idxptr, struct dn_list);
+	if (index_in_top_level == NULL) {
+		ldb_kv->idxptr->error = LDB_ERR_OPERATIONS_ERROR;
+		return -1;
+	}
+	index_in_top_level->dn
+		= talloc_steal(index_in_top_level,
+			       index_in_subtransaction->dn);
+	index_in_top_level->count = index_in_subtransaction->count;
+
+	rec.dptr = (uint8_t *)&index_in_top_level;
+	rec.dsize = sizeof(void *);
+
+
+	/*
+	 * This is not a store into the main DB, but into an in-memory
+	 * TDB, so we don't need a guard on ltdb->read_only
+	 */
+	ret = tdb_store(ldb_kv->idxptr->itdb, key, rec, TDB_INSERT);
+	if (ret != 0) {
+		ldb_kv->idxptr->error = ltdb_err_map(
+		    tdb_error(ldb_kv->idxptr->itdb));
+		return -1;
+	}
+	return 0;
+}
+
+/*
+ * Initialise the index cache for a sub transaction.
+ */
+int ldb_kv_index_sub_transaction_start(struct ldb_kv_private *ldb_kv)
+{
+	ldb_kv->nested_idx_ptr = talloc_zero(ldb_kv, struct ldb_kv_idxptr);
+	if (ldb_kv->nested_idx_ptr == NULL) {
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	/*
+	 * We use a tiny hash size for the sub-database (11).
+	 *
+	 * The sub-transaction is only for one record at a time, we
+	 * would use a linked list but that would make the code even
+	 * more complex when manipulating the index, as it would have
+	 * to know if we were in a nested transaction (normal
+	 * operations) or the top one (a reindex).
+	 */
+	ldb_kv->nested_idx_ptr->itdb =
+		tdb_open(NULL, 11, TDB_INTERNAL, O_RDWR, 0);
+	if (ldb_kv->nested_idx_ptr->itdb == NULL) {
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+	return LDB_SUCCESS;
+}
+
+/*
+ * Clear the contents of the nested transaction index cache when the nested
+ * transaction is cancelled.
+ */
+int ldb_kv_index_sub_transaction_cancel(struct ldb_kv_private *ldb_kv)
+{
+	if (ldb_kv->nested_idx_ptr != NULL) {
+		tdb_close(ldb_kv->nested_idx_ptr->itdb);
+		TALLOC_FREE(ldb_kv->nested_idx_ptr);
+	}
+	return LDB_SUCCESS;
+}
+
+/*
+ * Commit a nested transaction,
+ * Copy the contents of the nested transaction index cache to the
+ * transaction index cache.
+ */
+int ldb_kv_index_sub_transaction_commit(struct ldb_kv_private *ldb_kv)
+{
+	int ret = 0;
+
+	if (ldb_kv->nested_idx_ptr == NULL) {
+		return LDB_SUCCESS;
+	}
+	if (ldb_kv->nested_idx_ptr->itdb == NULL) {
+		return LDB_SUCCESS;
+	}
+	tdb_traverse(
+	    ldb_kv->nested_idx_ptr->itdb,
+	    ldb_kv_sub_transaction_traverse,
+	    ldb_kv->module);
+	tdb_close(ldb_kv->nested_idx_ptr->itdb);
+	ldb_kv->nested_idx_ptr->itdb = NULL;
+
+	ret = ldb_kv->nested_idx_ptr->error;
+	if (ret != LDB_SUCCESS) {
+		struct ldb_context *ldb = ldb_module_get_ctx(ldb_kv->module);
+		if (!ldb_errstring(ldb)) {
+			ldb_set_errstring(ldb, ldb_strerror(ret));
+		}
+		ldb_asprintf_errstring(
+			ldb,
+			__location__": Failed to update index records in "
+			"sub transaction commit: %s",
+			ldb_errstring(ldb));
+	}
+	TALLOC_FREE(ldb_kv->nested_idx_ptr);
+	return ret;
+}
diff --git a/lib/ldb/ldb_key_value/ldb_kv_search.c b/lib/ldb/ldb_key_value/ldb_kv_search.c
new file mode 100644
index 0000000..5dbbae6
--- /dev/null
+++ b/lib/ldb/ldb_key_value/ldb_kv_search.c
@@ -0,0 +1,795 @@
+/*
+   ldb database library
+
+   Copyright (C) Andrew Tridgell  2004
+
+     ** NOTE! The following LGPL license applies to the ldb
+     ** library. This does NOT imply that all of Samba is released
+     ** under the LGPL
+
+   This library is free software; you can redistribute it and/or
+   modify it under the terms of the GNU Lesser General Public
+   License as published by the Free Software Foundation; either
+   version 3 of the License, or (at your option) any later version.
+
+   This library is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+   Lesser General Public License for more details.
+
+   You should have received a copy of the GNU Lesser General Public
+   License along with this library; if not, see <http://www.gnu.org/licenses/>.
+*/
+
+/*
+ *  Name: ldb
+ *
+ *  Component: ldb search functions
+ *
+ *  Description: functions to search ldb+tdb databases
+ *
+ *  Author: Andrew Tridgell
+ */
+
+#include "ldb_kv.h"
+#include "ldb_private.h"
+#include "lib/util/attr.h"
+/*
+  search the database for a single simple dn.
+  return LDB_ERR_NO_SUCH_OBJECT on record-not-found
+  and LDB_SUCCESS on success
+*/
+int ldb_kv_search_base(struct ldb_module *module,
+		       TALLOC_CTX *mem_ctx,
+		       struct ldb_dn *dn,
+		       struct ldb_dn **ret_dn)
+{
+	int exists;
+	int ret;
+	struct ldb_message *msg = NULL;
+
+	if (ldb_dn_is_null(dn)) {
+		return LDB_ERR_NO_SUCH_OBJECT;
+	}
+
+	/*
+	 * We can't use tdb_exists() directly on a key when the TDB
+	 * key is the GUID one, not the DN based one.  So we just do a
+	 * normal search and avoid most of the allocation with the
+	 * LDB_UNPACK_DATA_FLAG_NO_ATTRS flag
+	 */
+	msg = ldb_msg_new(module);
+	if (msg == NULL) {
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	ret = ldb_kv_search_dn1(module, dn, msg, LDB_UNPACK_DATA_FLAG_NO_ATTRS);
+	if (ret == LDB_SUCCESS) {
+		const char *dn_linearized
+			= ldb_dn_get_linearized(dn);
+		const char *msg_dn_linearized
+			= ldb_dn_get_linearized(msg->dn);
+
+		if (strcmp(dn_linearized, msg_dn_linearized) == 0) {
+			/*
+			 * Re-use the full incoming DN for
+			 * subtree checks
+			 */
+			*ret_dn = dn;
+		} else {
+			/*
+			 * Use the string DN from the unpack, so that
+			 * we have a case-exact match of the base
+			 */
+			*ret_dn = talloc_steal(mem_ctx, msg->dn);
+		}
+		exists = true;
+	} else if (ret == LDB_ERR_NO_SUCH_OBJECT) {
+		exists = false;
+	} else {
+		talloc_free(msg);
+		return ret;
+	}
+	talloc_free(msg);
+	if (exists) {
+		return LDB_SUCCESS;
+	}
+	return LDB_ERR_NO_SUCH_OBJECT;
+}
+
+struct ldb_kv_parse_data_unpack_ctx {
+	struct ldb_message *msg;
+	struct ldb_module *module;
+	struct ldb_kv_private *ldb_kv;
+	unsigned int unpack_flags;
+};
+
+static int ldb_kv_parse_data_unpack(struct ldb_val key,
+				    struct ldb_val data,
+				    void *private_data)
+{
+	struct ldb_kv_parse_data_unpack_ctx *ctx = private_data;
+	int ret;
+	struct ldb_context *ldb = ldb_module_get_ctx(ctx->module);
+	struct ldb_val data_parse = data;
+
+	struct ldb_kv_private *ldb_kv = ctx->ldb_kv;
+
+	if ((ldb_kv->kv_ops->options & LDB_KV_OPTION_STABLE_READ_LOCK) &&
+	    (ctx->unpack_flags & LDB_UNPACK_DATA_FLAG_READ_LOCKED) &&
+	    !ldb_kv->kv_ops->transaction_active(ldb_kv)) {
+		/*
+		 * In the case where no transactions are active and
+		 * we're in a read-lock, we can point directly into
+		 * database memory.
+		 *
+		 * The database can't be changed underneath us and we
+		 * will duplicate this data in the call to filter.
+		 *
+		 * This is seen in:
+		 * - ldb_kv_index_filter
+		 * - ldb_kv_search_and_return_base
+		 */
+	} else {
+		/*
+		 * In every other case, since unpack doesn't memdup, we need
+		 * to at least do a memdup on the whole data buffer as that
+		 * may change later and the caller needs a stable result.
+		 *
+		 * During transactions, pointers could change and in
+		 * TDB, there just aren't the same guarantees.
+		 */
+		data_parse.data = talloc_memdup(ctx->msg,
+						data.data,
+						data.length);
+		if (data_parse.data == NULL) {
+			ldb_debug(ldb, LDB_DEBUG_ERROR,
+				  "Unable to allocate data(%d) for %*.*s\n",
+				  (int)data.length,
+				  (int)key.length, (int)key.length, key.data);
+			return LDB_ERR_OPERATIONS_ERROR;
+		}
+	}
+
+	ret = ldb_unpack_data_flags(ldb, &data_parse,
+				    ctx->msg, ctx->unpack_flags);
+	if (ret == -1) {
+		if (data_parse.data != data.data) {
+			talloc_free(data_parse.data);
+		}
+
+		ldb_debug(ldb, LDB_DEBUG_ERROR,
+			  __location__ ": Invalid data for index %*.*s\n",
+			  (int)key.length, (int)key.length, key.data);
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+	return ret;
+}
+
+/*
+  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 ldb_kv_search_key(struct ldb_module *module,
+		      struct ldb_kv_private *ldb_kv,
+		      const struct ldb_val ldb_key,
+		      struct ldb_message *msg,
+		      unsigned int unpack_flags)
+{
+	int ret;
+	struct ldb_kv_parse_data_unpack_ctx ctx = {
+		.msg = msg,
+		.module = module,
+		.unpack_flags = unpack_flags,
+		.ldb_kv = ldb_kv
+	};
+
+	memset(msg, 0, sizeof(*msg));
+
+	msg->num_elements = 0;
+	msg->elements = NULL;
+
+	ret = ldb_kv->kv_ops->fetch_and_parse(
+	    ldb_kv, ldb_key, ldb_kv_parse_data_unpack, &ctx);
+
+	if (ret == -1) {
+		ret = ldb_kv->kv_ops->error(ldb_kv);
+		if (ret == LDB_SUCCESS) {
+			/*
+			 * Just to be sure we don't turn errors
+			 * into success
+			 */
+			return LDB_ERR_OPERATIONS_ERROR;
+		}
+		return ret;
+	} else if (ret != LDB_SUCCESS) {
+		return ret;
+	}
+
+	return LDB_SUCCESS;
+}
+
+/*
+  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 ldb_kv_search_dn1(struct ldb_module *module,
+		      struct ldb_dn *dn,
+		      struct ldb_message *msg,
+		      unsigned int unpack_flags)
+{
+	void *data = ldb_module_get_private(module);
+	struct ldb_kv_private *ldb_kv =
+	    talloc_get_type(data, struct ldb_kv_private);
+	int ret;
+	uint8_t guid_key[LDB_KV_GUID_KEY_SIZE];
+	struct ldb_val key = {
+		.data = guid_key,
+		.length = sizeof(guid_key)
+	};
+	TALLOC_CTX *tdb_key_ctx = NULL;
+
+	bool valid_dn = ldb_dn_validate(dn);
+	if (valid_dn == false) {
+		ldb_asprintf_errstring(ldb_module_get_ctx(module),
+				       "Invalid Base DN: %s",
+				       ldb_dn_get_linearized(dn));
+		return LDB_ERR_INVALID_DN_SYNTAX;
+	}
+
+	if (ldb_kv->cache->GUID_index_attribute == NULL ||
+	    ldb_dn_is_special(dn)) {
+
+		tdb_key_ctx = talloc_new(msg);
+		if (!tdb_key_ctx) {
+			return ldb_module_oom(module);
+		}
+
+		/* form the key */
+		key = ldb_kv_key_dn(tdb_key_ctx, dn);
+		if (!key.data) {
+			TALLOC_FREE(tdb_key_ctx);
+			return LDB_ERR_OPERATIONS_ERROR;
+		}
+	} else {
+		/*
+		 * Look in the index to find the key for this DN.
+		 *
+		 * the tdb_key memory is allocated above, msg is just
+		 * used for internal memory.
+		 *
+		 */
+		ret = ldb_kv_key_dn_from_idx(module, ldb_kv, msg, dn, &key);
+		if (ret != LDB_SUCCESS) {
+			return ret;
+		}
+	}
+
+	ret = ldb_kv_search_key(module, ldb_kv, key, msg, unpack_flags);
+
+	TALLOC_FREE(tdb_key_ctx);
+
+	if (ret != LDB_SUCCESS) {
+		return ret;
+	}
+
+	if ((unpack_flags & LDB_UNPACK_DATA_FLAG_NO_DN) == 0) {
+		if (!msg->dn) {
+			msg->dn = ldb_dn_copy(msg, dn);
+		}
+		if (!msg->dn) {
+			return LDB_ERR_OPERATIONS_ERROR;
+		}
+	}
+
+	return LDB_SUCCESS;
+}
+
+/*
+ * filter the specified list of attributes from msg,
+ * adding requested attributes, and perhaps all for *.
+ * The DN will not be added if it is missing.
+ */
+int ldb_kv_filter_attrs_in_place(struct ldb_message *msg,
+				 const char *const *attrs)
+{
+	return ldb_filter_attrs_in_place(msg, attrs);
+}
+
+/*
+  search function for a non-indexed search
+ */
+static int search_func(_UNUSED_ struct ldb_kv_private *ldb_kv,
+		       struct ldb_val key,
+		       struct ldb_val val,
+		       void *state)
+{
+	struct ldb_context *ldb;
+	struct ldb_kv_context *ac;
+	struct ldb_message *msg;
+	struct timeval now;
+	int ret, timeval_cmp;
+	bool matched;
+
+	ac = talloc_get_type(state, struct ldb_kv_context);
+	ldb = ldb_module_get_ctx(ac->module);
+
+	/*
+	 * We want to skip @ records early in a search full scan
+	 *
+	 * @ records like @IDXLIST are only available via a base
+	 * search on the specific name but the method by which they
+	 * were excluded was expensive, after the unpack the DN is
+	 * exploded and ldb_match_msg_error() would reject it for
+	 * failing to match the scope.
+	 *
+	 * ldb_kv_key_is_normal_record() uses the fact that @ records
+	 * have the DN=@ prefix on their TDB/LMDB key to quickly
+	 * exclude them from consideration.
+	 *
+	 * (any other non-records are also excluded by the same key
+	 * match)
+	 */
+
+	if (ldb_kv_key_is_normal_record(key) == false) {
+		return 0;
+	}
+
+	/*
+	 * Check the time every 64 records, to reduce calls to
+	 * gettimeofday().  This is a compromise, not all calls to
+	 * ldb_match_message() will take the same time, most will fail
+	 * quickly but by luck it might be possible to have 64 records
+	 * that are slow, doing a recursive search via
+	 * LDAP_MATCHING_RULE_IN_CHAIN.
+	 */
+	if (ac->timeout_counter++ % 64 == 0) {
+		now = tevent_timeval_current();
+		timeval_cmp = tevent_timeval_compare(&ac->timeout_timeval,
+						     &now);
+
+		/*
+		 * The search has taken too long.  This is the most
+		 * likely place for our time to expire, as we are in
+		 * an un-indexed search and we return the data from
+		 * within this loop.  The tevent based timeout is not
+		 * likely to be hit, sadly.
+		 *
+		 * ldb_match_msg_error() can be quite expensive if a
+		 * LDAP_MATCHING_RULE_IN_CHAIN extended match was
+		 * specified.
+		 */
+		if (timeval_cmp <= 0) {
+			ac->error = LDB_ERR_TIME_LIMIT_EXCEEDED;
+			return -1;
+		}
+	}
+
+	msg = ldb_msg_new(ac);
+	if (!msg) {
+		ac->error = LDB_ERR_OPERATIONS_ERROR;
+		return -1;
+	}
+
+	/* unpack the record */
+	ret = ldb_unpack_data_flags(ldb, &val, msg,
+				    LDB_UNPACK_DATA_FLAG_NO_VALUES_ALLOC);
+	if (ret == -1) {
+		talloc_free(msg);
+		ac->error = LDB_ERR_OPERATIONS_ERROR;
+		return -1;
+	}
+
+	if (!msg->dn) {
+		msg->dn = ldb_dn_new(msg, ldb,
+				     (char *)key.data + 3);
+		if (msg->dn == NULL) {
+			talloc_free(msg);
+			ac->error = LDB_ERR_OPERATIONS_ERROR;
+			return -1;
+		}
+	}
+
+	/*
+	 * The redaction callback may be expensive to call if it fetches a
+	 * security descriptor. Check the DN early and bail out if it doesn't
+	 * match the base.
+	 */
+	if (!ldb_match_scope(ldb, ac->base, msg->dn, ac->scope)) {
+		talloc_free(msg);
+		return 0;
+	}
+
+	if (ldb->redact.callback != NULL) {
+		ret = ldb->redact.callback(ldb->redact.module, ac->req, msg);
+		if (ret != LDB_SUCCESS) {
+			talloc_free(msg);
+			return ret;
+		}
+	}
+
+	/* see if it matches the given expression */
+	ret = ldb_match_message(ldb, msg,
+				ac->tree, ac->scope, &matched);
+	if (ret != LDB_SUCCESS) {
+		talloc_free(msg);
+		ac->error = LDB_ERR_OPERATIONS_ERROR;
+		return -1;
+	}
+	if (!matched) {
+		talloc_free(msg);
+		return 0;
+	}
+
+	ret = ldb_msg_add_distinguished_name(msg);
+	if (ret == -1) {
+		talloc_free(msg);
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	/* filter the attributes that the user wants */
+	ret = ldb_kv_filter_attrs_in_place(msg, ac->attrs);
+	if (ret != LDB_SUCCESS) {
+		talloc_free(msg);
+		ac->error = LDB_ERR_OPERATIONS_ERROR;
+		return -1;
+	}
+
+	ldb_msg_shrink_to_fit(msg);
+
+	/* Ensure the message elements are all talloc'd. */
+	ret = ldb_msg_elements_take_ownership(msg);
+	if (ret != LDB_SUCCESS) {
+		talloc_free(msg);
+		ac->error = LDB_ERR_OPERATIONS_ERROR;
+		return -1;
+	}
+
+	ret = ldb_module_send_entry(ac->req, msg, NULL);
+	if (ret != LDB_SUCCESS) {
+		ac->request_terminated = true;
+		/* the callback failed, abort the operation */
+		ac->error = LDB_ERR_OPERATIONS_ERROR;
+		return -1;
+	}
+
+	return 0;
+}
+
+/*
+ * Key pointing to just before the first GUID indexed record for
+ * iterate_range
+ */
+struct ldb_val start_of_db_key = {.data=discard_const_p(uint8_t, "GUID<"),
+				  .length=6};
+/*
+ * Key pointing to just after the last GUID indexed record for
+ * iterate_range
+ */
+struct ldb_val end_of_db_key = {.data=discard_const_p(uint8_t, "GUID>"),
+				.length=6};
+
+/*
+  search the database with a LDAP-like expression.
+  this is the "full search" non-indexed variant
+*/
+static int ldb_kv_search_full(struct ldb_kv_context *ctx)
+{
+	void *data = ldb_module_get_private(ctx->module);
+	struct ldb_kv_private *ldb_kv =
+	    talloc_get_type(data, struct ldb_kv_private);
+	int ret;
+
+	/*
+	 * If the backend has an iterate_range op, use it to start the search
+	 * at the first GUID indexed record, skipping the indexes section.
+	 */
+	ctx->error = LDB_SUCCESS;
+	ret = ldb_kv->kv_ops->iterate_range(ldb_kv,
+					    start_of_db_key,
+					    end_of_db_key,
+					    search_func,
+					    ctx);
+	if (ret == LDB_ERR_OPERATIONS_ERROR) {
+		/*
+		 * If iterate_range isn't defined, it'll return an error,
+		 * so just iterate over the whole DB.
+		 */
+		ret = ldb_kv->kv_ops->iterate(ldb_kv, search_func, ctx);
+	}
+
+	if (ret < 0) {
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	return ctx->error;
+}
+
+static int ldb_kv_search_and_return_base(struct ldb_kv_private *ldb_kv,
+					 struct ldb_kv_context *ctx)
+{
+	struct ldb_message *msg;
+	struct ldb_context *ldb = ldb_module_get_ctx(ctx->module);
+	const char *dn_linearized;
+	const char *msg_dn_linearized;
+	int ret;
+	bool matched;
+
+	msg = ldb_msg_new(ctx);
+	if (!msg) {
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+	ret = ldb_kv_search_dn1(ctx->module,
+				ctx->base,
+				msg,
+				LDB_UNPACK_DATA_FLAG_NO_VALUES_ALLOC |
+				LDB_UNPACK_DATA_FLAG_READ_LOCKED);
+
+	if (ret == LDB_ERR_NO_SUCH_OBJECT) {
+		if (ldb_kv->check_base == false) {
+			/*
+			 * In this case, we are done, as no base
+			 * checking is allowed in this DB
+			 */
+			talloc_free(msg);
+			return LDB_SUCCESS;
+		}
+		ldb_asprintf_errstring(ldb,
+				       "No such Base DN: %s",
+				       ldb_dn_get_linearized(ctx->base));
+	}
+	if (ret != LDB_SUCCESS) {
+		talloc_free(msg);
+		return ret;
+	}
+
+	if (ldb->redact.callback != NULL) {
+		ret = ldb->redact.callback(ldb->redact.module, ctx->req, msg);
+		if (ret != LDB_SUCCESS) {
+			talloc_free(msg);
+			return ret;
+		}
+	}
+
+	/*
+	 * We use this, not ldb_match_msg_error() as we know
+	 * we matched on the scope BASE, as we just fetched
+	 * the base DN
+	 */
+
+	ret = ldb_match_message(ldb, msg,
+				ctx->tree,
+				ctx->scope,
+				&matched);
+	if (ret != LDB_SUCCESS) {
+		talloc_free(msg);
+		return ret;
+	}
+	if (!matched) {
+		talloc_free(msg);
+		return LDB_SUCCESS;
+	}
+
+	dn_linearized = ldb_dn_get_linearized(ctx->base);
+	msg_dn_linearized = ldb_dn_get_linearized(msg->dn);
+
+	if (strcmp(dn_linearized, msg_dn_linearized) == 0) {
+		/*
+		 * If the DN is exactly the same string, then
+		 * re-use the full incoming DN for the
+		 * returned result, as it has already been
+		 * casefolded
+		 */
+		struct ldb_dn *dn = ldb_dn_copy(msg, ctx->base);
+		if (dn != NULL) {
+			msg->dn = dn;
+		}
+	}
+
+	ret = ldb_msg_add_distinguished_name(msg);
+	if (ret == -1) {
+		talloc_free(msg);
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	/*
+	 * filter the attributes that the user wants.
+	 */
+	ret = ldb_kv_filter_attrs_in_place(msg, ctx->attrs);
+	if (ret != LDB_SUCCESS) {
+		talloc_free(msg);
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	ldb_msg_shrink_to_fit(msg);
+
+	/* Ensure the message elements are all talloc'd. */
+	ret = ldb_msg_elements_take_ownership(msg);
+	if (ret != LDB_SUCCESS) {
+		talloc_free(msg);
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	/*
+	 * Remove any extended components, we just want the casefold components
+	 */
+	ldb_dn_remove_extended_components(msg->dn);
+
+	ret = ldb_module_send_entry(ctx->req, msg, NULL);
+	if (ret != LDB_SUCCESS) {
+		/* Regardless of success or failure, the msg
+		 * is the callbacks responsibility, and should
+		 * not be talloc_free()'ed */
+		ctx->request_terminated = true;
+		return ret;
+	}
+
+	return LDB_SUCCESS;
+}
+
+/*
+  search the database with a LDAP-like expression.
+  choses a search method
+*/
+int ldb_kv_search(struct ldb_kv_context *ctx)
+{
+	struct ldb_context *ldb;
+	struct ldb_module *module = ctx->module;
+	struct ldb_request *req = ctx->req;
+	void *data = ldb_module_get_private(module);
+	struct ldb_kv_private *ldb_kv =
+	    talloc_get_type(data, struct ldb_kv_private);
+	int ret;
+
+	ldb = ldb_module_get_ctx(module);
+
+	ldb_request_set_state(req, LDB_ASYNC_PENDING);
+
+	if (ldb_kv->kv_ops->lock_read(module) != 0) {
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	if (ldb_kv_cache_load(module) != 0) {
+		ldb_kv->kv_ops->unlock_read(module);
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	if (req->op.search.tree == NULL) {
+		ldb_kv->kv_ops->unlock_read(module);
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	ctx->tree = req->op.search.tree;
+	ctx->scope = req->op.search.scope;
+	ctx->base = req->op.search.base;
+	ctx->attrs = req->op.search.attrs;
+
+	if ((req->op.search.base == NULL) || (ldb_dn_is_null(req->op.search.base) == true)) {
+
+		/* Check what we should do with a NULL dn */
+		switch (req->op.search.scope) {
+		case LDB_SCOPE_BASE:
+			ldb_asprintf_errstring(ldb,
+					       "NULL Base DN invalid for a base search");
+			ret = LDB_ERR_INVALID_DN_SYNTAX;
+			break;
+		case LDB_SCOPE_ONELEVEL:
+			ldb_asprintf_errstring(ldb,
+					       "NULL Base DN invalid for a one-level search");
+			ret = LDB_ERR_INVALID_DN_SYNTAX;
+			break;
+		case LDB_SCOPE_SUBTREE:
+		default:
+			/* We accept subtree searches from a NULL base DN, ie over the whole DB */
+			ret = LDB_SUCCESS;
+		}
+	} else if (req->op.search.scope == LDB_SCOPE_BASE) {
+
+		/*
+		 * If we are LDB_SCOPE_BASE, do just one search and
+		 * return early.  This is critical to ensure we do not
+		 * go into the index code for special DNs, as that
+		 * will try to look up an index record for a special
+		 * record (which doesn't exist).
+		 */
+		ret = ldb_kv_search_and_return_base(ldb_kv, ctx);
+
+		ldb_kv->kv_ops->unlock_read(module);
+
+		return ret;
+
+	} else if (ldb_kv->check_base) {
+		/*
+		 * This database has been marked as
+		 * 'checkBaseOnSearch', so do a spot check of the base
+		 * dn.  Also optimise the subsequent filter by filling
+		 * in the ctx->base to be exactly case correct
+		 */
+		ret = ldb_kv_search_base(
+		    module, ctx, req->op.search.base, &ctx->base);
+
+		if (ret == LDB_ERR_NO_SUCH_OBJECT) {
+			ldb_asprintf_errstring(ldb,
+					       "No such Base DN: %s",
+					       ldb_dn_get_linearized(req->op.search.base));
+		}
+
+	} else if (ldb_dn_validate(req->op.search.base) == false) {
+
+		/* We don't want invalid base DNs here */
+		ldb_asprintf_errstring(ldb,
+				       "Invalid Base DN: %s",
+				       ldb_dn_get_linearized(req->op.search.base));
+		ret = LDB_ERR_INVALID_DN_SYNTAX;
+
+	} else {
+		/* If we are not checking the base DN life is easy */
+		ret = LDB_SUCCESS;
+	}
+
+	if (ret == LDB_SUCCESS) {
+		uint32_t match_count = 0;
+
+		ret = ldb_kv_search_indexed(ctx, &match_count);
+		if (ret == LDB_ERR_NO_SUCH_OBJECT) {
+			/* Not in the index, therefore OK! */
+			ret = LDB_SUCCESS;
+
+		}
+		/* Check if we got just a normal error.
+		 * In that case proceed to a full search unless we got a
+		 * callback error */
+		if (!ctx->request_terminated && ret != LDB_SUCCESS) {
+			/* Not indexed, so we need to do a full scan */
+			if (ldb_kv->warn_unindexed ||
+			    ldb_kv->disable_full_db_scan) {
+				/* useful for debugging when slow performance
+				 * is caused by unindexed searches */
+				char *expression = ldb_filter_from_tree(ctx, ctx->tree);
+				ldb_debug(ldb, LDB_DEBUG_ERROR, "ldb FULL SEARCH: %s SCOPE: %s DN: %s",
+							expression,
+							req->op.search.scope==LDB_SCOPE_BASE?"base":
+							req->op.search.scope==LDB_SCOPE_ONELEVEL?"one":
+							req->op.search.scope==LDB_SCOPE_SUBTREE?"sub":"UNKNOWN",
+							ldb_dn_get_linearized(req->op.search.base));
+
+				talloc_free(expression);
+			}
+
+			if (match_count != 0) {
+				/* the indexing code gave an error
+				 * after having returned at least one
+				 * entry. This means the indexes are
+				 * corrupt or a database record is
+				 * corrupt. We cannot continue with a
+				 * full search or we may return
+				 * duplicate entries
+				 */
+				ldb_kv->kv_ops->unlock_read(module);
+				return LDB_ERR_OPERATIONS_ERROR;
+			}
+
+			if (ldb_kv->disable_full_db_scan) {
+				ldb_set_errstring(ldb,
+						  "ldb FULL SEARCH disabled");
+				ldb_kv->kv_ops->unlock_read(module);
+				return LDB_ERR_INAPPROPRIATE_MATCHING;
+			}
+
+			ret = ldb_kv_search_full(ctx);
+			if (ret != LDB_SUCCESS) {
+				ldb_set_errstring(ldb, "Indexed and full searches both failed!\n");
+			}
+		}
+	}
+
+	ldb_kv->kv_ops->unlock_read(module);
+
+	return ret;
+}