--- a/plugins/mod_storage_sql.lua	Wed Mar 11 14:36:56 2020 +0000
+++ b/plugins/mod_storage_sql.lua	Wed Mar 11 15:57:53 2020 +0000
@@ -230,6 +230,50 @@
 	return result;
 end
 
+function map_store:find_key(key)
+	if type(key) ~= "string" or key == "" then
+		return nil, "find_key only supports non-empty string keys";
+	end
+	local ok, result = engine:transaction(function()
+		local query = [[
+		SELECT "user", "type", "value"
+		FROM "prosody"
+		WHERE "host"=? AND "store"=? AND "key"=?
+		]];
+
+		local data;
+		for row in engine:select(query, host, self.store, key) do
+			local key_data, err = deserialize(row[2], row[3]);
+			assert(key_data ~= nil, err);
+			if data == nil then
+				data = {};
+			end
+			data[row[1]] = key_data;
+		end
+
+		return data;
+
+	end);
+	if not ok then return nil, result; end
+	return result;
+end
+
+function map_store:delete_key(key)
+	if type(key) ~= "string" or key == "" then
+		return nil, "delete_key only supports non-empty string keys";
+	end
+	local ok, result = engine:transaction(function()
+		local delete_sql = [[
+		DELETE FROM "prosody"
+		WHERE "host"=? AND "store"=? AND "key"=?;
+		]];
+		engine:delete(delete_sql, host, self.store, key);
+		return true;
+	end);
+	if not ok then return nil, result; end
+	return result;
+end
+
 local archive_store = {}
 archive_store.caps = {
 	total = true;

--- a/spec/core_storagemanager_spec.lua	Wed Mar 11 14:36:56 2020 +0000
+++ b/spec/core_storagemanager_spec.lua	Wed Mar 11 15:57:53 2020 +0000
@@ -62,6 +62,8 @@
 			sm.initialize_host(test_host);
 			assert(mm.load(test_host, "storage_"..backend_config.storage));
 
+			local sql_it = backend_config.sql and it or pending;
+
 			describe("key-value stores", function ()
 				-- These tests rely on being executed in order, disable any order
 				-- randomization for this block
@@ -113,15 +115,83 @@
 					assert.equal("bar", store:get("user9999", "foo"));
 				end);
 
-				it("may remove data for a user", function ()
+				sql_it("may find all users with a specific key", function ()
+					assert.is_function(store.find_key);
+					assert(store:set("user9999b", "bar", "bar"));
+					assert(store:set("user9999c", "foo", "blah"));
+					local ret, err = store:find_key("foo");
+					assert.is_nil(err);
+					assert.same({ user9999 = "bar", user9999c = "blah" }, ret);
+				end);
+
+				sql_it("rejects empty or non-string keys to find_key", function ()
+					assert.is_function(store.find_key);
+					do
+						local ret, err = store:find_key("");
+						assert.is_nil(ret);
+						assert.is_not_nil(err);
+					end
+					do
+						local ret, err = store:find_key(true);
+						assert.is_nil(ret);
+						assert.is_not_nil(err);
+					end
+				end);
+
+				sql_it("rejects empty or non-string keys to delete_key", function ()
+					assert.is_function(store.delete_key);
+					do
+						local ret, err = store:delete_key("");
+						assert.is_nil(ret);
+						assert.is_not_nil(err);
+					end
+					do
+						local ret, err = store:delete_key(true);
+						assert.is_nil(ret);
+						assert.is_not_nil(err);
+					end
+				end);
+
+				sql_it("may delete all instances of a specific key", function ()
+					assert.is_function(store.delete_key);
+					assert(store:set("user9999b", "foo", "hello"));
+
+					local ret, err = store:delete_key("bar");
+					-- Ensure key was deleted
+					do
+						local ret, err = store:get("user9999b", "bar");
+						assert.is_nil(ret);
+						assert.is_nil(err);
+					end
+					-- Ensure other users/keys are intact
+					do
+						local ret, err = store:get("user9999", "foo");
+						assert.equal("bar", ret);
+						assert.is_nil(err);
+					end
+					do
+						local ret, err = store:get("user9999b", "foo");
+						assert.equal("hello", ret);
+						assert.is_nil(err);
+					end
+					do
+						local ret, err = store:get("user9999c", "foo");
+						assert.equal("blah", ret);
+						assert.is_nil(err);
+					end
+				end);
+
+				it("may remove data for a specific key for a user", function ()
 					assert(store:set("user9999", "foo", nil));
 					do
 						local ret, err = store:get("user9999", "foo");
 						assert.is_nil(ret);
 						assert.is_nil(err);
 					end
+
+					assert(store:set("user9999b", "foo", nil));
 					do
-						local ret, err = kv_store:get("user9999");
+						local ret, err = store:get("user9999b", "foo");
 						assert.is_nil(ret);
 						assert.is_nil(err);
 					end