int partition_reload_if_required(struct ldb_module *module, 
				 struct partition_private_data *data,
				 struct ldb_request *parent)
{
	uint64_t seq;
	int ret;
	unsigned int i;
	struct ldb_context *ldb = ldb_module_get_ctx(module);
	struct ldb_message *msg;
	struct ldb_message_element *partition_attributes;
	struct ldb_message_element *partial_replicas;
	TALLOC_CTX *mem_ctx;

	if (!data) {
		/* Not initialised yet */
		return LDB_SUCCESS;
	}

	mem_ctx = talloc_new(data);
	if (!mem_ctx) {
		return ldb_oom(ldb);
	}

	ret = partition_primary_sequence_number(module, mem_ctx, &seq, parent);
	if (ret != LDB_SUCCESS) {
		talloc_free(mem_ctx);
		return ret;
	}
	if (seq == data->metadata_seq) {
		talloc_free(mem_ctx);
		return LDB_SUCCESS;
	}

	/* This loads metadata tdb. If it's missing, creates it */
	ret = partition_metadata_init(module);
	if (ret != LDB_SUCCESS) {
		return ret;
	}

	ret = partition_reload_metadata(module, data, mem_ctx, &msg, parent);
	if (ret != LDB_SUCCESS) {
		talloc_free(mem_ctx);
		return ret;
	}

	data->metadata_seq = seq;

	partition_attributes = ldb_msg_find_element(msg, "partition");
	partial_replicas     = ldb_msg_find_element(msg, "partialReplica");
	data->backend_db_store
		= talloc_strdup(data, ldb_msg_find_attr_as_string(msg, "backendStore", "tdb"));

	if (data->backend_db_store == NULL) {
		talloc_free(mem_ctx);
		return ldb_module_oom(module);
	}

	for (i=0; partition_attributes && i < partition_attributes->num_values; i++) {
		unsigned int j;
		bool new_partition = true;
		const char *filename = NULL;
		DATA_BLOB dn_blob;
		struct ldb_dn *dn;
		struct dsdb_partition *partition;
		struct ldb_result *dn_res;
		const char *no_attrs[] = { NULL };

		for (j=0; data->partitions && data->partitions[j]; j++) {
			if (data_blob_cmp(&data->partitions[j]->orig_record, &partition_attributes->values[i]) == 0) {
				new_partition = false;
				break;
			}
		}
		if (new_partition == false) {
			continue;
		}

		dn_blob = partition_attributes->values[i];
		
		if (dn_blob.length > 4 && 
		    (strncmp((const char *)&dn_blob.data[dn_blob.length-4], ".ldb", 4) == 0)) {

			/* Look for DN:filename.ldb */
			char *p = strrchr((const char *)dn_blob.data, ':');
			if (!p) {
				ldb_asprintf_errstring(ldb, 
						       "partition_init: invalid DN in attempting to parse partition record: %s", (const char *)dn_blob.data);
				talloc_free(mem_ctx);
				return LDB_ERR_CONSTRAINT_VIOLATION;
			}
			filename = p+1;
			
			/* Now trim off the filename */
			dn_blob.length = ((uint8_t *)p - dn_blob.data);
		}

		dn = ldb_dn_from_ldb_val(mem_ctx, ldb, &dn_blob);
		if (!dn) {
			ldb_asprintf_errstring(ldb, 
					       "partition_init: invalid DN in partition record: %s", (const char *)dn_blob.data);
			talloc_free(mem_ctx);
			return LDB_ERR_CONSTRAINT_VIOLATION;
		}
		
		/* Now do a slow check with the DN compare */
		for (j=0; data->partitions && data->partitions[j]; j++) {
			if (ldb_dn_compare(dn, data->partitions[j]->ctrl->dn) == 0) {
				new_partition = false;
				break;
			}
		}
		if (new_partition == false) {
			continue;
		}

		if (!filename) {
			char *base64_dn = NULL;
			const char *p;
			for (p = ldb_dn_get_linearized(dn); *p; p++) {
				/* We have such a strict check because I don't want shell metacharacters in the file name, nor ../ */
				if (!(isalnum(*p) || *p == ' ' || *p == '=' || *p == ',')) {
					break;
				}
			}
			if (*p) {
				base64_dn = ldb_base64_encode(data, ldb_dn_get_linearized(dn), strlen(ldb_dn_get_linearized(dn)));
				filename = talloc_asprintf(mem_ctx, "%s.ldb", base64_dn);
			} else {
				filename = talloc_asprintf(mem_ctx, "%s.ldb", ldb_dn_get_linearized(dn));
			}
		}
			
		/* We call ldb_dn_get_linearized() because the DN in
		 * partition_attributes is already casefolded
		 * correctly.  We don't want to mess that up as the
		 * schema isn't loaded yet */
		ret = new_partition_from_dn(ldb, data, data->partitions, dn, 
					    filename, data->backend_db_store,
					    &partition);
		if (ret != LDB_SUCCESS) {
			talloc_free(mem_ctx);
			return ret;
		}

		talloc_steal(partition, partition_attributes->values[i].data);
		partition->orig_record = partition_attributes->values[i];

		/* Get the 'correct' case of the partition DNs from the database */
		ret = dsdb_module_search_dn(partition->module, data, &dn_res, 
					    dn, no_attrs,
					    DSDB_FLAG_NEXT_MODULE, parent);
		if (ret == LDB_SUCCESS) {
			talloc_free(partition->ctrl->dn);
			partition->ctrl->dn = talloc_steal(partition->ctrl, dn_res->msgs[0]->dn);
			talloc_free(dn_res);
		} else if (ret != LDB_ERR_NO_SUCH_OBJECT) {
			ldb_asprintf_errstring(ldb,
					       "Failed to search for partition base %s in new partition at %s: %s", 
					       ldb_dn_get_linearized(dn), 
					       partition->backend_url, 
					       ldb_errstring(ldb));
			talloc_free(mem_ctx);
			return ret;
		}

		/* see if it is a partial replica */
		for (j=0; partial_replicas && j<partial_replicas->num_values; j++) {
			struct ldb_dn *pa_dn = ldb_dn_from_ldb_val(mem_ctx, ldb, &partial_replicas->values[j]);
			if (pa_dn != NULL && ldb_dn_compare(pa_dn, partition->ctrl->dn) == 0) {
				partition->partial_replica = true;
			}
			talloc_free(pa_dn);
		}

		ret = add_partition_to_data(ldb, data, partition);
		if (ret != LDB_SUCCESS) {
			talloc_free(mem_ctx);
			return ret;
		}
	}

	talloc_free(mem_ctx);
	return LDB_SUCCESS;
}