[datastore] Fix #3743

This change adds support for key == NULL to the datastore plugins
and replaces the offset argument with a next_uid and random arguments to
increase performance in the key == NULL case.

With the offset argument a datastore plugin would have to count all
matching keys before fetching the key at the right offset, which would
iterate over the entire database in the case of key == NULL.

The offset argument was used in two ways: to iterate over a set of
matching values and to start iteration at a random matching value. The new API
seperates these into two arguments: if random is true it will return a
random matching value, otherwise next_uid can be set to uid + 1 to return the
next matching value.

The random argument was not added to get_zero_anonymity. This function
is used to periodically insert zero anonymity values into the DHT. I
don't think it's necessary to randomize this.

1 files changed, 48 insertions, 213 deletions

diff --git a/src/datastore/plugin_datastore_sqlite.c b/src/datastore/plugin_datastore_sqlite.c
index 9ca8f056a..76f791ad4 100644
--- a/src/datastore/plugin_datastore_sqlite.c
+++ b/src/datastore/plugin_datastore_sqlite.c
@@ -130,42 +130,7 @@ struct Plugin
   /**
    * Precompiled SQL for selection
    */
-  sqlite3_stmt *count_key;
-
-  /**
-   * Precompiled SQL for selection
-   */
-  sqlite3_stmt *count_key_vhash;
-
-  /**
-   * Precompiled SQL for selection
-   */
-  sqlite3_stmt *count_key_type;
-
-  /**
-   * Precompiled SQL for selection
-   */
-  sqlite3_stmt *count_key_vhash_type;
-
-  /**
-   * Precompiled SQL for selection
-   */
-  sqlite3_stmt *get_key;
-
-  /**
-   * Precompiled SQL for selection
-   */
-  sqlite3_stmt *get_key_vhash;
-
-  /**
-   * Precompiled SQL for selection
-   */
-  sqlite3_stmt *get_key_type;
-
-  /**
-   * Precompiled SQL for selection
-   */
-  sqlite3_stmt *get_key_vhash_type;
+  sqlite3_stmt *get;
 
   /**
    * Should the database be dropped on shutdown?
@@ -430,8 +395,10 @@ database_setup (const struct GNUNET_CONFIGURATION_Handle *cfg,
 #if SQLITE_VERSION_NUMBER >= 3007000
                     "INDEXED BY idx_anon_type_hash "
 #endif
-                    "WHERE (anonLevel = 0 AND type=?1) "
-                    "ORDER BY hash DESC LIMIT 1 OFFSET ?2",
+                    "WHERE _ROWID_ >= ? AND "
+                    "anonLevel = 0 AND "
+                    "type = ? "
+                    "ORDER BY _ROWID_ ASC LIMIT 1",
                     &plugin->selZeroAnon)) ||
        (SQLITE_OK !=
         sq_prepare (plugin->dbh,
@@ -440,44 +407,14 @@ database_setup (const struct GNUNET_CONFIGURATION_Handle *cfg,
                     &plugin->insertContent)) ||
        (SQLITE_OK !=
         sq_prepare (plugin->dbh,
-                    "SELECT count(*) FROM gn090 WHERE hash=?",
-                    &plugin->count_key)) ||
-       (SQLITE_OK !=
-        sq_prepare (plugin->dbh,
-                    "SELECT count(*) FROM gn090 WHERE hash=? AND vhash=?",
-                    &plugin->count_key_vhash)) ||
-       (SQLITE_OK !=
-        sq_prepare (plugin->dbh,
-                    "SELECT count(*) FROM gn090 WHERE hash=? AND type=?",
-                    &plugin->count_key_type)) ||
-       (SQLITE_OK !=
-        sq_prepare (plugin->dbh,
-                    "SELECT count(*) FROM gn090 WHERE hash=? AND vhash=? AND type=?",
-                    &plugin->count_key_vhash_type)) ||
-       (SQLITE_OK !=
-        sq_prepare (plugin->dbh,
-                    "SELECT type, prio, anonLevel, expire, hash, value, _ROWID_ FROM gn090 "
-                    "WHERE hash=?"
-                    "ORDER BY _ROWID_ ASC LIMIT 1 OFFSET ?",
-                    &plugin->get_key)) ||
-       (SQLITE_OK !=
-        sq_prepare (plugin->dbh,
-                    "SELECT type, prio, anonLevel, expire, hash, value, _ROWID_ FROM gn090 "
-                    "WHERE hash=? AND vhash=?"
-                    "ORDER BY _ROWID_ ASC LIMIT 1 OFFSET ?",
-                    &plugin->get_key_vhash)) ||
-       (SQLITE_OK !=
-        sq_prepare (plugin->dbh,
                     "SELECT type, prio, anonLevel, expire, hash, value, _ROWID_ FROM gn090 "
-                    "WHERE hash=? AND type=?"
-                    "ORDER BY _ROWID_ ASC LIMIT 1 OFFSET ?",
-                    &plugin->get_key_type)) ||
-       (SQLITE_OK !=
-        sq_prepare (plugin->dbh,
-                    "SELECT type, prio, anonLevel, expire, hash, value, _ROWID_ FROM gn090 "
-                    "WHERE hash=? AND vhash=? AND type=?"
-                    "ORDER BY _ROWID_ ASC LIMIT 1 OFFSET ?",
-                    &plugin->get_key_vhash_type)) ||
+                    "WHERE _ROWID_ >= ? AND "
+                    "(rvalue >= ? OR 0 = ?) AND "
+                    "(hash = ? OR 0 = ?) AND "
+                    "(vhash = ? OR 0 = ?) AND "
+                    "(type = ? OR 0 = ?) "
+                    "ORDER BY _ROWID_ ASC LIMIT 1",
+                    &plugin->get)) ||
        (SQLITE_OK !=
         sq_prepare (plugin->dbh,
                     "DELETE FROM gn090 WHERE _ROWID_ = ?",
@@ -523,22 +460,8 @@ database_shutdown (struct Plugin *plugin)
     sqlite3_finalize (plugin->selZeroAnon);
   if (NULL != plugin->insertContent)
     sqlite3_finalize (plugin->insertContent);
-  if (NULL != plugin->count_key)
-    sqlite3_finalize (plugin->count_key);
-  if (NULL != plugin->count_key_vhash)
-    sqlite3_finalize (plugin->count_key_vhash);
-  if (NULL != plugin->count_key_type)
-    sqlite3_finalize (plugin->count_key_type);
-  if (NULL != plugin->count_key_vhash_type)
-    sqlite3_finalize (plugin->count_key_vhash_type);
-  if (NULL != plugin->count_key)
-    sqlite3_finalize (plugin->get_key);
-  if (NULL != plugin->count_key_vhash)
-    sqlite3_finalize (plugin->get_key_vhash);
-  if (NULL != plugin->count_key_type)
-    sqlite3_finalize (plugin->get_key_type);
-  if (NULL != plugin->count_key_vhash_type)
-    sqlite3_finalize (plugin->get_key_vhash_type);
+  if (NULL != plugin->get)
+    sqlite3_finalize (plugin->get);
   result = sqlite3_close (plugin->dbh);
 #if SQLITE_VERSION_NUMBER >= 3007000
   if (result == SQLITE_BUSY)
@@ -895,38 +818,36 @@ execute_get (struct Plugin *plugin,
  * the given processor for the item.
  *
  * @param cls our plugin context
- * @param offset offset of the result (modulo num-results);
- *               specific ordering does not matter for the offset
+ * @param next_uid return the result with lowest uid >= next_uid
  * @param type entries of which type should be considered?
- *        Use 0 for any type.
- * @param proc function to call on each matching value;
- *        will be called once with a NULL value at the end
+ *        Must not be zero (ANY).
+ * @param proc function to call on the matching value;
+ *        will be called with NULL if no value matches
  * @param proc_cls closure for @a proc
  */
 static void
 sqlite_plugin_get_zero_anonymity (void *cls,
-                                  uint64_t offset,
+                                  uint64_t next_uid,
                                   enum GNUNET_BLOCK_Type type,
                                   PluginDatumProcessor proc,
                                   void *proc_cls)
 {
   struct Plugin *plugin = cls;
   struct GNUNET_SQ_QueryParam params[] = {
+    GNUNET_SQ_query_param_uint64 (&next_uid),
     GNUNET_SQ_query_param_uint32 (&type),
-    GNUNET_SQ_query_param_uint64 (&offset),
     GNUNET_SQ_query_param_end
   };
-  sqlite3_stmt *stmt = plugin->selZeroAnon;
 
   GNUNET_assert (type != GNUNET_BLOCK_TYPE_ANY);
   if (GNUNET_OK !=
-      GNUNET_SQ_bind (stmt,
+      GNUNET_SQ_bind (plugin->selZeroAnon,
                       params))
   {
     proc (proc_cls, NULL, 0, NULL, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
     return;
   }
-  execute_get (plugin, stmt, proc, proc_cls);
+  execute_get (plugin, plugin->selZeroAnon, proc, proc_cls);
 }
 
 
@@ -934,8 +855,9 @@ sqlite_plugin_get_zero_anonymity (void *cls,
  * Get results for a particular key in the datastore.
  *
  * @param cls closure
- * @param offset offset (mod count).
- * @param key key to match, never NULL
+ * @param next_uid return the result with lowest uid >= next_uid
+ * @param random if true, return a random result instead of using next_uid
+ * @param key maybe NULL (to match all entries)
  * @param vhash hash of the value, maybe NULL (to
  *        match all values that have the right key).
  *        Note that for DBlocks there is no difference
@@ -949,7 +871,8 @@ sqlite_plugin_get_zero_anonymity (void *cls,
  */
 static void
 sqlite_plugin_get_key (void *cls,
-                       uint64_t offset,
+                       uint64_t next_uid,
+                       bool random,
                        const struct GNUNET_HashCode *key,
                        const struct GNUNET_HashCode *vhash,
                        enum GNUNET_BLOCK_Type type,
@@ -957,133 +880,45 @@ sqlite_plugin_get_key (void *cls,
                        void *proc_cls)
 {
   struct Plugin *plugin = cls;
+  uint64_t rvalue;
+  uint16_t use_rvalue = random;
   uint32_t type32 = (uint32_t) type;
-  int ret;
-  int total;
-  uint32_t limit_off;
-  struct GNUNET_SQ_QueryParam count_params_key[] = {
-    GNUNET_SQ_query_param_auto_from_type (key),
-    GNUNET_SQ_query_param_end
-  };
-  struct GNUNET_SQ_QueryParam count_params_key_vhash[] = {
-    GNUNET_SQ_query_param_auto_from_type (key),
-    GNUNET_SQ_query_param_auto_from_type (vhash),
-    GNUNET_SQ_query_param_end
-  };
-  struct GNUNET_SQ_QueryParam count_params_key_type[] = {
-    GNUNET_SQ_query_param_auto_from_type (key),
-    GNUNET_SQ_query_param_uint32 (&type32),
-    GNUNET_SQ_query_param_end
-  };
-  struct GNUNET_SQ_QueryParam count_params_key_vhash_type[] = {
-    GNUNET_SQ_query_param_auto_from_type (key),
-    GNUNET_SQ_query_param_auto_from_type (vhash),
-    GNUNET_SQ_query_param_uint32 (&type32),
-    GNUNET_SQ_query_param_end
-  };
-  struct GNUNET_SQ_QueryParam get_params_key[] = {
-    GNUNET_SQ_query_param_auto_from_type (key),
-    GNUNET_SQ_query_param_uint32 (&limit_off),
-    GNUNET_SQ_query_param_end
-  };
-  struct GNUNET_SQ_QueryParam get_params_key_vhash[] = {
-    GNUNET_SQ_query_param_auto_from_type (key),
-    GNUNET_SQ_query_param_auto_from_type (vhash),
-    GNUNET_SQ_query_param_uint32 (&limit_off),
-    GNUNET_SQ_query_param_end
-  };
-  struct GNUNET_SQ_QueryParam get_params_key_type[] = {
-    GNUNET_SQ_query_param_auto_from_type (key),
-    GNUNET_SQ_query_param_uint32 (&type32),
-    GNUNET_SQ_query_param_uint32 (&limit_off),
-    GNUNET_SQ_query_param_end
-  };
-  struct GNUNET_SQ_QueryParam get_params_key_vhash_type[] = {
+  uint16_t use_type = GNUNET_BLOCK_TYPE_ANY != type;
+  uint16_t use_key = NULL != key;
+  uint16_t use_vhash = NULL != vhash;
+  struct GNUNET_SQ_QueryParam params[] = {
+    GNUNET_SQ_query_param_uint64 (&next_uid),
+    GNUNET_SQ_query_param_uint64 (&rvalue),
+    GNUNET_SQ_query_param_uint16 (&use_rvalue),
     GNUNET_SQ_query_param_auto_from_type (key),
+    GNUNET_SQ_query_param_uint16 (&use_key),
     GNUNET_SQ_query_param_auto_from_type (vhash),
+    GNUNET_SQ_query_param_uint16 (&use_vhash),
     GNUNET_SQ_query_param_uint32 (&type32),
-    GNUNET_SQ_query_param_uint32 (&limit_off),
+    GNUNET_SQ_query_param_uint16 (&use_type),
     GNUNET_SQ_query_param_end
   };
-  struct GNUNET_SQ_QueryParam *count_params;
-  sqlite3_stmt *count_stmt;
-  struct GNUNET_SQ_QueryParam *get_params;
-  sqlite3_stmt *get_stmt;
 
-  if (NULL == vhash)
+  if (random)
   {
-    if (GNUNET_BLOCK_TYPE_ANY == type)
-    {
-      count_params = count_params_key;
-      count_stmt = plugin->count_key;
-      get_params = get_params_key;
-      get_stmt = plugin->get_key;
-    }
-    else
-    {
-      count_params = count_params_key_type;
-      count_stmt = plugin->count_key_type;
-      get_params = get_params_key_type;
-      get_stmt = plugin->get_key_type;
-    }
+    rvalue = GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_WEAK,
+                                       UINT64_MAX);
+    next_uid = 0;
   }
   else
-  {
-    if (GNUNET_BLOCK_TYPE_ANY == type)
-    {
-      count_params = count_params_key_vhash;
-      count_stmt = plugin->count_key_vhash;
-      get_params = get_params_key_vhash;
-      get_stmt = plugin->get_key_vhash;
-    }
-    else
-    {
-      count_params = count_params_key_vhash_type;
-      count_stmt = plugin->count_key_vhash_type;
-      get_params = get_params_key_vhash_type;
-      get_stmt = plugin->get_key_vhash_type;
-    }
-  }
-  if (GNUNET_OK !=
-      GNUNET_SQ_bind (count_stmt,
-                      count_params))
-  {
-    proc (proc_cls, NULL, 0, NULL, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
-    return;
-  }
-  ret = sqlite3_step (count_stmt);
-  if (ret != SQLITE_ROW)
-  {
-    LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
-                "sqlite_step");
-    GNUNET_SQ_reset (plugin->dbh,
-                     count_stmt);
-    proc (proc_cls, NULL, 0, NULL, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
-    return;
-  }
-  total = sqlite3_column_int (count_stmt,
-                              0);
-  GNUNET_SQ_reset (plugin->dbh,
-                   count_stmt);
-  if (0 == total)
-  {
-    proc (proc_cls, NULL, 0, NULL, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
-    return;
-  }
-  limit_off = (uint32_t) (offset % total);
+    rvalue = 0;
+
   if (GNUNET_OK !=
-      GNUNET_SQ_bind (get_stmt,
-                      get_params))
+      GNUNET_SQ_bind (plugin->get,
+                      params))
   {
     proc (proc_cls, NULL, 0, NULL, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
     return;
   }
   execute_get (plugin,
-               get_stmt,
+               plugin->get,
                proc,
                proc_cls);
-  GNUNET_SQ_reset (plugin->dbh,
-                   get_stmt);
 }