ldb_index.c

samba最新软件

/* 
   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 tdb backend - indexing
 *
 *  Description: indexing routines for ldb tdb backend
 *
 *  Author: Andrew Tridgell
 */

#include "ldb_includes.h"

#include "ldb_tdb.h"

/*
  find an element in a list, using the given comparison function and
  assuming that the list is already sorted using comp_fn

  return -1 if not found, or the index of the first occurance of needle if found
*/
static int ldb_list_find(const void *needle, 
			 const void *base, size_t nmemb, size_t size, 
			 comparison_fn_t comp_fn)
{
	const char *base_p = (const char *)base;
	size_t min_i, max_i, test_i;

	if (nmemb == 0) {
		return -1;
	}

	min_i = 0;
	max_i = nmemb-1;

	while (min_i < max_i) {
		int r;

		test_i = (min_i + max_i) / 2;
		/* the following cast looks strange, but is
		 correct. The key to understanding it is that base_p
		 is a pointer to an array of pointers, so we have to
		 dereference it after casting to void **. The strange
		 const in the middle gives us the right type of pointer
		 after the dereference  (tridge) */
		r = comp_fn(needle, *(void * const *)(base_p + (size * test_i)));
		if (r == 0) {
			/* scan back for first element */
			while (test_i > 0 &&
			       comp_fn(needle, *(void * const *)(base_p + (size * (test_i-1)))) == 0) {
				test_i--;
			}
			return test_i;
		}
		if (r < 0) {
			if (test_i == 0) {
				return -1;
			}
			max_i = test_i - 1;
		}
		if (r > 0) {
			min_i = test_i + 1;
		}
	}

	if (comp_fn(needle, *(void * const *)(base_p + (size * min_i))) == 0) {
		return min_i;
	}

	return -1;
}

struct dn_list {
	unsigned int count;
	char **dn;
};

/*
  return the dn key to be used for an index
  caller frees
*/
static struct ldb_dn *ltdb_index_key(struct ldb_context *ldb,
				     const char *attr, const struct ldb_val *value)
{
	struct ldb_dn *ret;
	struct ldb_val v;
	const struct ldb_schema_attribute *a;
	char *attr_folded;
	int r;

	attr_folded = ldb_attr_casefold(ldb, attr);
	if (!attr_folded) {
		return NULL;
	}

	a = ldb_schema_attribute_by_name(ldb, attr);
	r = a->syntax->canonicalise_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;
	}
	if (ldb_should_b64_encode(&v)) {
		char *vstr = ldb_base64_encode(ldb, (char *)v.data, v.length);
		if (!vstr) return NULL;
		ret = ldb_dn_new_fmt(ldb, ldb, "%s:%s::%s", LTDB_INDEX, attr_folded, vstr);
		talloc_free(vstr);
	} else {
		ret = ldb_dn_new_fmt(ldb, ldb, "%s:%s:%.*s", LTDB_INDEX, attr_folded, (int)v.length, (char *)v.data);
	}

	if (v.data != value->data) {
		talloc_free(v.data);
	}
	talloc_free(attr_folded);

	return ret;
}

/*
  see if a attribute value is in the list of indexed attributes
*/
static int ldb_msg_find_idx(const struct ldb_message *msg, const char *attr,
			    unsigned int *v_idx, const char *key)
{
	unsigned int i, j;
	for (i=0;i<msg->num_elements;i++) {
		if (ldb_attr_cmp(msg->elements[i].name, key) == 0) {
			const struct ldb_message_element *el = &msg->elements[i];

			if (attr == NULL) {
				/* in this case we are just looking to see if key is present,
 				   we are not spearching for a specific index */
				return 0;
			}

			for (j=0;j<el->num_values;j++) {
				if (ldb_attr_cmp((char *)el->values[j].data, attr) == 0) {
					if (v_idx) {
						*v_idx = j;
					}
					return i;
				}
			}
		}
	}
	return -1;
}

/* used in sorting dn lists */
static int list_cmp(const char **s1, const char **s2)
{
	return strcmp(*s1, *s2);
}

/*
  return a list of dn's that might match a simple indexed search or
 */
static int ltdb_index_dn_simple(struct ldb_module *module, 
				const struct ldb_parse_tree *tree,
				const struct ldb_message *index_list,
				struct dn_list *list)
{
	struct ldb_context *ldb = module->ldb;
	struct ldb_dn *dn;
	int ret;
	unsigned int i, j;
	struct ldb_message *msg;

	list->count = 0;
	list->dn = NULL;

	/* if the attribute isn't in the list of indexed attributes then
	   this node needs a full search */
	if (ldb_msg_find_idx(index_list, tree->u.equality.attr, NULL, LTDB_IDXATTR) == -1) {
		return LDB_ERR_OPERATIONS_ERROR;
	}

	/* the attribute is indexed. Pull the list of DNs that match the 
	   search criterion */
	dn = ltdb_index_key(ldb, tree->u.equality.attr, &tree->u.equality.value);
	if (!dn) return LDB_ERR_OPERATIONS_ERROR;

	msg = talloc(list, struct ldb_message);
	if (msg == NULL) {
		return LDB_ERR_OPERATIONS_ERROR;
	}

	ret = ltdb_search_dn1(module, dn, msg);
	talloc_free(dn);
	if (ret != LDB_SUCCESS) {
		return ret;
	}

	for (i=0;i<msg->num_elements;i++) {
		struct ldb_message_element *el;

		if (strcmp(msg->elements[i].name, LTDB_IDX) != 0) {
			continue;
		}

		el = &msg->elements[i];

		list->dn = talloc_array(list, char *, el->num_values);
		if (!list->dn) {
			talloc_free(msg);
			return LDB_ERR_OPERATIONS_ERROR;
		}

		for (j=0;j<el->num_values;j++) {
			list->dn[list->count] = 
				talloc_strdup(list->dn, (char *)el->values[j].data);
			if (!list->dn[list->count]) {
				talloc_free(msg);
				return LDB_ERR_OPERATIONS_ERROR;
			}
			list->count++;
		}
	}

	talloc_free(msg);

	if (list->count > 1) {
		qsort(list->dn, list->count, sizeof(char *), (comparison_fn_t) list_cmp);
	}

	return LDB_SUCCESS;
}


static int list_union(struct ldb_context *, struct dn_list *, const struct dn_list *);

/*
  return a list of dn's that might match a leaf indexed search
 */
static int ltdb_index_dn_leaf(struct ldb_module *module, 
			      const struct ldb_parse_tree *tree,
			      const struct ldb_message *index_list,
			      struct dn_list *list)
{
	if (ldb_attr_dn(tree->u.equality.attr) == 0) {
		list->dn = talloc_array(list, char *, 1);
		if (list->dn == NULL) {
			ldb_oom(module->ldb);
			return LDB_ERR_OPERATIONS_ERROR;
		}
		list->dn[0] = talloc_strdup(list->dn, (char *)tree->u.equality.value.data);
		if (list->dn[0] == NULL) {
			ldb_oom(module->ldb);
			return LDB_ERR_OPERATIONS_ERROR;
		}
		list->count = 1;
		return LDB_SUCCESS;
	}
	return ltdb_index_dn_simple(module, tree, index_list, list);
}


/*
  list intersection
  list = list & list2
  relies on the lists being sorted
*/
static int list_intersect(struct ldb_context *ldb,
			  struct dn_list *list, const struct dn_list *list2)
{
	struct dn_list *list3;
	unsigned int i;

	if (list->count == 0 || list2->count == 0) {
		/* 0 & X == 0 */
		return LDB_ERR_NO_SUCH_OBJECT;
	}

	list3 = talloc(ldb, struct dn_list);
	if (list3 == NULL) {
		return LDB_ERR_OPERATIONS_ERROR;
	}

	list3->dn = talloc_array(list3, char *, list->count);
	if (!list3->dn) {
		talloc_free(list3);
		return LDB_ERR_OPERATIONS_ERROR;
	}
	list3->count = 0;

	for (i=0;i<list->count;i++) {
		if (ldb_list_find(list->dn[i], list2->dn, list2->count, 
			      sizeof(char *), (comparison_fn_t)strcmp) != -1) {
			list3->dn[list3->count] = talloc_move(list3->dn, &list->dn[i]);
			list3->count++;
		} else {
			talloc_free(list->dn[i]);
		}		
	}

	talloc_free(list->dn);
	list->dn = talloc_move(list, &list3->dn);
	list->count = list3->count;
	talloc_free(list3);

	return LDB_ERR_NO_SUCH_OBJECT;
}


/*
  list union
  list = list | list2
  relies on the lists being sorted
*/
static int list_union(struct ldb_context *ldb, 
		      struct dn_list *list, const struct dn_list *list2)
{
	unsigned int i;
	char **d;
	unsigned int count = list->count;

	if (list->count == 0 && list2->count == 0) {
		/* 0 | 0 == 0 */
		return LDB_ERR_NO_SUCH_OBJECT;
	}

	d = talloc_realloc(list, list->dn, char *, list->count + list2->count);
	if (!d) {
		return LDB_ERR_OPERATIONS_ERROR;
	}
	list->dn = d;

	for (i=0;i<list2->count;i++) {
		if (ldb_list_find(list2->dn[i], list->dn, count, 
			      sizeof(char *), (comparison_fn_t)strcmp) == -1) {
			list->dn[list->count] = talloc_strdup(list->dn, list2->dn[i]);
			if (!list->dn[list->count]) {
				return LDB_ERR_OPERATIONS_ERROR;
			}
			list->count++;
		}		
	}

	if (list->count != count) {
		qsort(list->dn, list->count, sizeof(char *), (comparison_fn_t)list_cmp);
	}

	return LDB_ERR_NO_SUCH_OBJECT;
}

static int ltdb_index_dn(struct ldb_module *module, 
			 const struct ldb_parse_tree *tree,
			 const struct ldb_message *index_list,
			 struct dn_list *list);


/*
  OR two index results
 */
static int ltdb_index_dn_or(struct ldb_module *module, 
			    const struct ldb_parse_tree *tree,
			    const struct ldb_message *index_list,
			    struct dn_list *list)
{
	struct ldb_context *ldb = module->ldb;
	unsigned int i;
	int ret;
	
	ret = LDB_ERR_OPERATIONS_ERROR;
	list->dn = NULL;
	list->count = 0;

	for (i=0;i<tree->u.list.num_elements;i++) {
		struct dn_list *list2;
		int v;

		list2 = talloc(module, struct dn_list);
		if (list2 == NULL) {
			return LDB_ERR_OPERATIONS_ERROR;
		}

		v = ltdb_index_dn(module, tree->u.list.elements[i], index_list, list2);

		if (v == LDB_ERR_NO_SUCH_OBJECT) {
			/* 0 || X == X */
			if (ret != LDB_SUCCESS && ret != LDB_ERR_NO_SUCH_OBJECT) {
				ret = v;
			}
			talloc_free(list2);
			continue;
		}

		if (v != LDB_SUCCESS && v != LDB_ERR_NO_SUCH_OBJECT) {
			/* 1 || X == 1 */
			talloc_free(list->dn);
			talloc_free(list2);
			return v;
		}

		if (ret != LDB_SUCCESS && ret != LDB_ERR_NO_SUCH_OBJECT) {
			ret = LDB_SUCCESS;
			list->dn = talloc_move(list, &list2->dn);
			list->count = list2->count;
		} else {
			if (list_union(ldb, list, list2) == -1) {
				talloc_free(list2);
				return LDB_ERR_OPERATIONS_ERROR;
			}
			ret = LDB_SUCCESS;
		}
		talloc_free(list2);
	}

	if (list->count == 0) {
		return LDB_ERR_NO_SUCH_OBJECT;
	}

	return ret;
}


/*
  NOT an index results
 */
static int ltdb_index_dn_not(struct ldb_module *module, 
			     const struct ldb_parse_tree *tree,
			     const struct ldb_message *index_list,
			     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;
}

/*
  AND two index results
 */
static int ltdb_index_dn_and(struct ldb_module *module, 
			     const struct ldb_parse_tree *tree,
			     const struct ldb_message *index_list,
			     struct dn_list *list)
{
	struct ldb_context *ldb = module->ldb;
	unsigned int i;
	int ret;
	
	ret = LDB_ERR_OPERATIONS_ERROR;
	list->dn = NULL;
	list->count = 0;

	for (i=0;i<tree->u.list.num_elements;i++) {
		struct dn_list *list2;
		int v;

		list2 = talloc(module, struct dn_list);
		if (list2 == NULL) {
			return LDB_ERR_OPERATIONS_ERROR;
		}

		v = ltdb_index_dn(module, tree->u.list.elements[i], index_list, list2);

		if (v == LDB_ERR_NO_SUCH_OBJECT) {
			/* 0 && X == 0 */
			talloc_free(list->dn);
			talloc_free(list2);
			return LDB_ERR_NO_SUCH_OBJECT;
		}

		if (v != LDB_SUCCESS && v != LDB_ERR_NO_SUCH_OBJECT) {
			talloc_free(list2);
			continue;
		}

		if (ret != LDB_SUCCESS && ret != LDB_ERR_NO_SUCH_OBJECT) {
			ret = LDB_SUCCESS;
			talloc_free(list->dn);
			list->dn = talloc_move(list, &list2->dn);
			list->count = list2->count;
		} else {
			if (list_intersect(ldb, list, list2) == -1) {
				talloc_free(list2);
				return LDB_ERR_OPERATIONS_ERROR;
			}
		}

		talloc_free(list2);

		if (list->count == 0) {
			talloc_free(list->dn);
			return LDB_ERR_NO_SUCH_OBJECT;
		}
	}

	return ret;
}

/*
  AND index results and ONE level special index
 */
static int ltdb_index_dn_one(struct ldb_module *module, 
			     struct ldb_dn *parent_dn,
			     struct dn_list *list)
{
	struct ldb_context *ldb = module->ldb;
	struct dn_list *list2;
	struct ldb_message *msg;
	struct ldb_dn *key;
	struct ldb_val val;
	unsigned int i, j;
	int ret;
	
	list2 = talloc_zero(module, struct dn_list);
	if (list2 == NULL) {
		return LDB_ERR_OPERATIONS_ERROR;
	}

	/* the attribute is indexed. Pull the list of DNs that match the 
	   search criterion */
	val.data = (uint8_t *)((uintptr_t)ldb_dn_get_casefold(parent_dn));
	val.length = strlen((char *)val.data);
	key = ltdb_index_key(ldb, LTDB_IDXONE, &val);
	if (!key) {
		talloc_free(list2);
		return LDB_ERR_OPERATIONS_ERROR;
	}

	msg = talloc(list2, struct ldb_message);
	if (msg == NULL) {
		talloc_free(list2);
		return LDB_ERR_OPERATIONS_ERROR;
	}

	ret = ltdb_search_dn1(module, key, msg);
	talloc_free(key);
	if (ret != LDB_SUCCESS) {
		return ret;
	}

	for (i = 0; i < msg->num_elements; i++) {
		struct ldb_message_element *el;

		if (strcmp(msg->elements[i].name, LTDB_IDX) != 0) {
			continue;
		}

		el = &msg->elements[i];

		list2->dn = talloc_array(list2, char *, el->num_values);
		if (!list2->dn) {
			talloc_free(list2);
			return LDB_ERR_OPERATIONS_ERROR;
		}

		for (j = 0; j < el->num_values; j++) {
			list2->dn[list2->count] = talloc_strdup(list2->dn, (char *)el->values[j].data);
			if (!list2->dn[list2->count]) {
				talloc_free(list2);
				return LDB_ERR_OPERATIONS_ERROR;
			}
			list2->count++;
		}
	}

	if (list2->count == 0) {
		talloc_free(list2);
		return LDB_ERR_NO_SUCH_OBJECT;
	}

	if (list2->count > 1) {
		qsort(list2->dn, list2->count, sizeof(char *), (comparison_fn_t) list_cmp);
	}

	if (list->count > 0) {
		if (list_intersect(ldb, list, list2) == -1) {
			talloc_free(list2);
			return LDB_ERR_OPERATIONS_ERROR;
		}

		if (list->count == 0) {
			talloc_free(list->dn);
			talloc_free(list2);
			return LDB_ERR_NO_SUCH_OBJECT;
		}
	} else {
		list->dn = talloc_move(list, &list2->dn);
		list->count = list2->count;
	}

	talloc_free(list2);

	return LDB_SUCCESS;
}

/*
  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 ltdb_index_dn(struct ldb_module *module, 
			 const struct ldb_parse_tree *tree,
			 const struct ldb_message *index_list,
			 struct dn_list *list)
{
	int ret = LDB_ERR_OPERATIONS_ERROR;

	switch (tree->operation) {
	case LDB_OP_AND:
		ret = ltdb_index_dn_and(module, tree, index_list, list);
		break;