ldb_kv: Avoid memdup of database records in the case of base searches

[sfrench/samba-autobuild/.git] / lib / ldb / ldb_key_value / ldb_kv_search.c

diff --git a/lib/ldb/ldb_key_value/ldb_kv_search.c b/lib/ldb/ldb_key_value/ldb_kv_search.c
index e9964c2bd63b95ed0939d13ade5c476e8df3d169..a2946d6506b0392315b47defd3a98878becd1325 100644 (file)
--- a/lib/ldb/ldb_key_value/ldb_kv_search.c
+++ b/lib/ldb/ldb_key_value/ldb_kv_search.c
@@ -190,22 +190,45 @@ static int ldb_kv_parse_data_unpack(struct ldb_val key,
        struct ldb_context *ldb = ldb_module_get_ctx(ctx->module);
        struct ldb_val data_parse = data;
 
-       if (ctx->unpack_flags & LDB_UNPACK_DATA_FLAG_NO_DATA_ALLOC) {
-               /*
-                * If we got LDB_UNPACK_DATA_FLAG_NO_DATA_ALLOC
-                * we need at least do a memdup on the whole
-                * data buffer as that may change later
-                * and the caller needs a stable result.
-                */
-               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;
+       struct ldb_kv_private *ldb_kv =
+           talloc_get_type(ldb_module_get_private(ctx->module), struct ldb_kv_private);
+
+       if ((ctx->unpack_flags & LDB_UNPACK_DATA_FLAG_NO_DATA_ALLOC)) {
+               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, if we got
+                        * LDB_UNPACK_DATA_FLAG_NO_DATA_ALLOC we need 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;
+                       }
                }
        }
 
@@ -512,7 +535,24 @@ static int search_func(struct ldb_kv_private *ldb_kv,
        ac = talloc_get_type(state, struct ldb_kv_context);
        ldb = ldb_module_get_ctx(ac->module);
 
-       if (ldb_kv_key_is_record(key) == false) {
+       /*
+        * 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;
        }
 
@@ -618,7 +658,8 @@ static int ldb_kv_search_and_return_base(struct ldb_kv_private *ldb_kv,
                                ctx->base,
                                msg,
                                LDB_UNPACK_DATA_FLAG_NO_DATA_ALLOC |
-                                   LDB_UNPACK_DATA_FLAG_NO_VALUES_ALLOC);
+                               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) {
@@ -678,6 +719,11 @@ static int ldb_kv_search_and_return_base(struct ldb_kv_private *ldb_kv,
         * assignment is safe
         */
        ret = ldb_kv_filter_attrs(ctx, msg, ctx->attrs, &filtered_msg);
+       if (ret == -1) {
+               talloc_free(msg);
+               filtered_msg = NULL;
+               return LDB_ERR_OPERATIONS_ERROR;
+       }
 
        /*
         * Remove any extended components possibly copied in from
@@ -686,10 +732,6 @@ static int ldb_kv_search_and_return_base(struct ldb_kv_private *ldb_kv,
        ldb_dn_remove_extended_components(filtered_msg->dn);
        talloc_free(msg);
 
-       if (ret == -1) {
-               return LDB_ERR_OPERATIONS_ERROR;
-       }
-
        ret = ldb_module_send_entry(ctx->req, filtered_msg, NULL);
        if (ret != LDB_SUCCESS) {
                /* Regardless of success or failure, the msg
@@ -758,14 +800,6 @@ int ldb_kv_search(struct ldb_kv_context *ctx)
                        /* We accept subtree searches from a NULL base DN, ie over the whole DB */
                        ret = LDB_SUCCESS;
                }
-       } else if (ldb_dn_is_valid(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 (req->op.search.scope == LDB_SCOPE_BASE) {
 
                /*