diff --git a/src/spider/CMakeLists.txt b/src/spider/CMakeLists.txt
index d550ca11b..f8db97dbb 100644
--- a/src/spider/CMakeLists.txt
+++ b/src/spider/CMakeLists.txt
@@ -10,6 +10,7 @@ set(SPIDER_CORE_SOURCES
 set(SPIDER_CORE_HEADERS
     core/Error.hpp
     core/Data.hpp
+    core/KeyValueData.hpp
     core/Task.hpp
     core/TaskGraph.hpp
     core/JobMetadata.hpp
@@ -17,6 +18,7 @@ set(SPIDER_CORE_HEADERS
     io/MsgPack.hpp
     io/msgpack_message.hpp
     io/Serializer.hpp
+    utils/TimedCache.hpp
     storage/MetadataStorage.hpp
     storage/DataStorage.hpp
     storage/MysqlStorage.hpp
diff --git a/src/spider/core/Data.hpp b/src/spider/core/Data.hpp
index 40100e4e6..2ca23f8ae 100644
--- a/src/spider/core/Data.hpp
+++ b/src/spider/core/Data.hpp
@@ -1,7 +1,6 @@
 #ifndef SPIDER_CORE_DATA_HPP
 #define SPIDER_CORE_DATA_HPP
 
-#include <optional>
 #include <string>
 #include <utility>
 #include <vector>
@@ -9,9 +8,6 @@
 #include <boost/uuid/random_generator.hpp>
 #include <boost/uuid/uuid.hpp>
 
-#include "../io/MsgPack.hpp"  // IWYU pragma: keep
-#include "../io/Serializer.hpp"  // IWYU pragma: keep
-
 namespace spider::core {
 class Data {
 public:
@@ -19,25 +15,12 @@ class Data {
 
     explicit Data(std::string value) : m_value(std::move(value)) { init_id(); }
 
-    Data(boost::uuids::uuid id, std::string value) : m_id(id), m_value(std::move(value)) {}
-
-    Data(std::string key, std::string value) : m_key(std::move(key)), m_value(std::move(value)) {
-        init_id();
-    }
-
-    Data(boost::uuids::uuid id, std::string key, std::string value)
-            : m_id(id),
-              m_key(std::move(key)),
-              m_value(std::move(value)) {}
-
-    MSGPACK_DEFINE(m_id, m_key, m_value, m_locality, m_hard_locality);
+    Data(boost::uuids::uuid const id, std::string value) : m_id(id), m_value(std::move(value)) {}
 
     static auto is_data() -> bool { return true; }
 
     [[nodiscard]] auto get_id() const -> boost::uuids::uuid { return m_id; }
 
-    [[nodiscard]] auto get_key() const -> std::optional<std::string> { return m_key; }
-
     [[nodiscard]] auto get_value() const -> std::string { return m_value; }
 
     [[nodiscard]] auto get_locality() const -> std::vector<std::string> const& {
@@ -48,11 +31,10 @@ class Data {
 
     void set_locality(std::vector<std::string> const& locality) { m_locality = locality; }
 
-    void set_hard_locality(bool hard) { m_hard_locality = hard; }
+    void set_hard_locality(bool const hard) { m_hard_locality = hard; }
 
 private:
     boost::uuids::uuid m_id;
-    std::optional<std::string> m_key;
     std::string m_value;
     std::vector<std::string> m_locality;
     bool m_hard_locality = false;
diff --git a/src/spider/core/KeyValueData.hpp b/src/spider/core/KeyValueData.hpp
new file mode 100644
index 000000000..611f974f3
--- /dev/null
+++ b/src/spider/core/KeyValueData.hpp
@@ -0,0 +1,30 @@
+#ifndef SPIDER_CORE_KEYVALUEDATA_HPP
+#define SPIDER_CORE_KEYVALUEDATA_HPP
+
+#include <string>
+#include <utility>
+
+#include <boost/uuid/uuid.hpp>
+
+namespace spider::core {
+class KeyValueData {
+public:
+    KeyValueData(std::string key, std::string value, boost::uuids::uuid const id)
+            : m_key{std::move(key)},
+              m_value{std::move(value)},
+              m_id{id} {}
+
+    [[nodiscard]] auto get_key() const -> std::string const& { return m_key; }
+
+    [[nodiscard]] auto get_value() const -> std::string const& { return m_value; }
+
+    [[nodiscard]] auto get_id() const -> boost::uuids::uuid const& { return m_id; }
+
+private:
+    std::string m_key;
+    std::string m_value;
+    boost::uuids::uuid m_id;
+};
+}  // namespace spider::core
+
+#endif  // SPIDER_CORE_KEYVALUEDATA_HPP
diff --git a/src/spider/scheduler/FifoPolicy.cpp b/src/spider/scheduler/FifoPolicy.cpp
index 2f1bf8e80..b17e5ed08 100644
--- a/src/spider/scheduler/FifoPolicy.cpp
+++ b/src/spider/scheduler/FifoPolicy.cpp
@@ -8,7 +8,6 @@
 #include <string>
 #include <vector>
 
-#include <absl/container/flat_hash_map.h>
 #include <boost/uuid/uuid.hpp>
 #include <boost/uuid/uuid_io.hpp>
 #include <fmt/format.h>
@@ -83,8 +82,10 @@ auto FifoPolicy::schedule_next(
              metadata_store](core::Task const& task) -> std::chrono::system_clock::time_point {
                 boost::uuids::uuid const task_id = task.get_id();
                 boost::uuids::uuid job_id;
-                if (m_task_job_map.contains(task_id)) {
-                    job_id = m_task_job_map[task_id];
+                std::optional<boost::uuids::uuid> const optional_job_id
+                        = m_task_job_cache.get(task_id);
+                if (optional_job_id.has_value()) {
+                    job_id = optional_job_id.value();
                 } else {
                     if (false == metadata_store->get_task_job_id(task_id, &job_id).success()) {
                         throw std::runtime_error(fmt::format(
@@ -92,11 +93,13 @@ auto FifoPolicy::schedule_next(
                                 boost::uuids::to_string(task_id)
                         ));
                     }
-                    m_task_job_map.emplace(task_id, job_id);
+                    m_task_job_cache.put(task_id, job_id);
                 }
 
-                if (m_job_time_map.contains(job_id)) {
-                    return m_job_time_map[job_id];
+                std::optional<std::chrono::system_clock::time_point> const optional_time
+                        = m_job_time_cache.get(job_id);
+                if (optional_time.has_value()) {
+                    return optional_time.value();
                 }
 
                 core::JobMetadata job_metadata;
@@ -106,7 +109,7 @@ auto FifoPolicy::schedule_next(
                             boost::uuids::to_string(job_id)
                     ));
                 }
-                m_job_time_map.emplace(job_id, job_metadata.get_creation_time());
+                m_job_time_cache.put(job_id, job_metadata.get_creation_time());
                 return job_metadata.get_creation_time();
             }
     );
@@ -114,12 +117,9 @@ auto FifoPolicy::schedule_next(
     return earliest_task->get_id();
 }
 
-auto FifoPolicy::cleanup_job(boost::uuids::uuid const job_id) -> void {
-    absl::erase_if(m_task_job_map, [&job_id](auto const& item) -> bool {
-        auto const& [item_task_id, item_job_id] = item;
-        return item_job_id == job_id;
-    });
-    m_job_time_map.erase(job_id);
+auto FifoPolicy::cleanup() -> void {
+    m_task_job_cache.cleanup();
+    m_job_time_cache.cleanup();
 }
 
 }  // namespace spider::scheduler
diff --git a/src/spider/scheduler/FifoPolicy.hpp b/src/spider/scheduler/FifoPolicy.hpp
index 0ee1f1321..c1d32042b 100644
--- a/src/spider/scheduler/FifoPolicy.hpp
+++ b/src/spider/scheduler/FifoPolicy.hpp
@@ -6,11 +6,11 @@
 #include <optional>
 #include <string>
 
-#include <absl/container/flat_hash_map.h>
 #include <boost/uuid/uuid.hpp>
 
 #include "../storage/DataStorage.hpp"
 #include "../storage/MetadataStorage.hpp"
+#include "../utils/TimedCache.hpp"
 #include "SchedulerPolicy.hpp"
 
 namespace spider::scheduler {
@@ -23,11 +23,11 @@ class FifoPolicy final : public SchedulerPolicy {
             boost::uuids::uuid worker_id,
             std::string const& worker_addr
     ) -> std::optional<boost::uuids::uuid> override;
-    auto cleanup_job(boost::uuids::uuid job_id) -> void override;
+    auto cleanup() -> void override;
 
 private:
-    absl::flat_hash_map<boost::uuids::uuid, boost::uuids::uuid> m_task_job_map;
-    absl::flat_hash_map<boost::uuids::uuid, std::chrono::system_clock::time_point> m_job_time_map;
+    core::TimedCache<boost::uuids::uuid, boost::uuids::uuid> m_task_job_cache;
+    core::TimedCache<boost::uuids::uuid, std::chrono::system_clock::time_point> m_job_time_cache;
 };
 
 }  // namespace spider::scheduler
diff --git a/src/spider/scheduler/SchedulerPolicy.hpp b/src/spider/scheduler/SchedulerPolicy.hpp
index 8295975e4..2eb9f8503 100644
--- a/src/spider/scheduler/SchedulerPolicy.hpp
+++ b/src/spider/scheduler/SchedulerPolicy.hpp
@@ -27,7 +27,7 @@ class SchedulerPolicy {
             std::string const& worker_addr
     ) -> std::optional<boost::uuids::uuid> = 0;
 
-    virtual auto cleanup_job(boost::uuids::uuid job_id) -> void = 0;
+    virtual auto cleanup() -> void = 0;
 };
 
 }  // namespace spider::scheduler
diff --git a/src/spider/storage/DataStorage.hpp b/src/spider/storage/DataStorage.hpp
index 7c6394122..29be26e54 100644
--- a/src/spider/storage/DataStorage.hpp
+++ b/src/spider/storage/DataStorage.hpp
@@ -7,6 +7,7 @@
 
 #include "../core/Data.hpp"
 #include "../core/Error.hpp"
+#include "../core/KeyValueData.hpp"
 
 namespace spider::core {
 class DataStorage {
@@ -22,9 +23,9 @@ class DataStorage {
     virtual void close() = 0;
     virtual auto initialize() -> StorageErr = 0;
 
-    virtual auto add_data(Data const& data) -> StorageErr = 0;
+    virtual auto add_driver_data(boost::uuids::uuid driver_id, Data const& data) -> StorageErr = 0;
+    virtual auto add_task_data(boost::uuids::uuid task_id, Data const& data) -> StorageErr = 0;
     virtual auto get_data(boost::uuids::uuid id, Data* data) -> StorageErr = 0;
-    virtual auto get_data_by_key(std::string const& key, Data* data) -> StorageErr = 0;
     virtual auto remove_data(boost::uuids::uuid id) -> StorageErr = 0;
     virtual auto add_task_reference(boost::uuids::uuid id, boost::uuids::uuid task_id) -> StorageErr
                                                                                           = 0;
@@ -34,6 +35,20 @@ class DataStorage {
     add_driver_reference(boost::uuids::uuid id, boost::uuids::uuid driver_id) -> StorageErr = 0;
     virtual auto
     remove_driver_reference(boost::uuids::uuid id, boost::uuids::uuid driver_id) -> StorageErr = 0;
+    virtual auto remove_dangling_data() -> StorageErr = 0;
+
+    virtual auto add_client_kv_data(KeyValueData const& data) -> StorageErr = 0;
+    virtual auto add_task_kv_data(KeyValueData const& data) -> StorageErr = 0;
+    virtual auto get_client_kv_data(
+            boost::uuids::uuid const& client_id,
+            std::string const& key,
+            std::string* value
+    ) -> StorageErr = 0;
+    virtual auto get_task_kv_data(
+            boost::uuids::uuid const& task_id,
+            std::string const& key,
+            std::string* value
+    ) -> StorageErr = 0;
 };
 }  // namespace spider::core
 
diff --git a/src/spider/storage/MysqlStorage.cpp b/src/spider/storage/MysqlStorage.cpp
index 213f2f4ab..729b4763b 100644
--- a/src/spider/storage/MysqlStorage.cpp
+++ b/src/spider/storage/MysqlStorage.cpp
@@ -26,11 +26,11 @@
 #include <mariadb/conncpp/Properties.hpp>
 #include <mariadb/conncpp/ResultSet.hpp>
 #include <mariadb/conncpp/Statement.hpp>
-#include <mariadb/conncpp/Types.hpp>
 
 #include "../core/Data.hpp"
 #include "../core/Error.hpp"
 #include "../core/JobMetadata.hpp"
+#include "../core/KeyValueData.hpp"
 #include "../core/Task.hpp"
 #include "../core/TaskGraph.hpp"
 
@@ -127,12 +127,9 @@ char const* const cCreateTaskInstanceTable = R"(CREATE TABLE IF NOT EXISTS `task
 
 char const* const cCreateDataTable = R"(CREATE TABLE IF NOT EXISTS `data` (
     `id` BINARY(16) NOT NULL,
-    `key` VARCHAR(64),
     `value` VARCHAR(256) NOT NULL,
     `hard_locality` BOOL DEFAULT FALSE,
-    `gc` BOOL DEFAULT FALSE,
     `persisted` BOOL DEFAULT FALSE,
-    UNIQUE KEY (`key`) USING BTREE,
     PRIMARY KEY (`id`)
 ))";
 
@@ -161,7 +158,22 @@ char const* const cCreateDataRefTaskTable = R"(CREATE TABLE IF NOT EXISTS `data_
     CONSTRAINT `data_ref_task_id` FOREIGN KEY (`task_id`) REFERENCES `tasks` (`id`) ON UPDATE NO ACTION ON DELETE CASCADE
 ))";
 
-std::array<char const* const, 12> const cCreateStorage = {
+char const* const cCreateClientKVDataTable = R"(CREATE TABLE IF NOT EXISTS `client_kv_data` (
+    `kv_key` VARCHAR(64) NOT NULL,
+    `value` VARCHAR(128) NOT NULL,
+    `client_id` BINARY(16) NOT NULL,
+    PRIMARY KEY (`client_id`, `kv_key`)
+))";
+
+char const* const cCreateTaskKVDataTable = R"(CREATE TABLE IF NOT EXISTS `task_kv_data` (
+    `kv_key` VARCHAR(64) NOT NULL,
+    `value` VARCHAR(128) NOT NULL,
+    `task_id` BINARY(16) NOT NULL,
+    PRIMARY KEY (`task_id`, `kv_key`),
+    CONSTRAINT `kv_data_task_id` FOREIGN KEY (`task_id`) REFERENCES `tasks` (`id`) ON UPDATE NO ACTION ON DELETE CASCADE
+))";
+
+std::array<char const* const, 14> const cCreateStorage = {
         cCreateDriverTable,  // drivers table must be created before data_ref_driver
         cCreateSchedulerTable,
         cCreateJobTable,  // jobs table must be created before task
@@ -170,6 +182,8 @@ std::array<char const* const, 12> const cCreateStorage = {
         cCreateDataLocalityTable,
         cCreateDataRefDriverTable,
         cCreateDataRefTaskTable,
+        cCreateClientKVDataTable,
+        cCreateTaskKVDataTable,
         cCreateTaskOutputTable,  // task_outputs table must be created before task_inputs
         cCreateTaskInputTable,
         cCreateTaskDependencyTable,
@@ -1192,22 +1206,16 @@ auto MySqlDataStorage::initialize() -> StorageErr {
     return StorageErr{};
 }
 
-auto MySqlDataStorage::add_data(Data const& data) -> StorageErr {
+auto MySqlDataStorage::add_driver_data(boost::uuids::uuid const driver_id, Data const& data)
+        -> StorageErr {
     try {
-        std::unique_ptr<sql::PreparedStatement> statement(
-                m_conn->prepareStatement("INSERT INTO `data` (`id`, `key`, `value`, "
-                                         "`hard_locality`) VALUES(?, ?, ?, ?)")
-        );
+        std::unique_ptr<sql::PreparedStatement> statement(m_conn->prepareStatement(
+                "INSERT INTO `data` (`id`, `value`, `hard_locality`) VALUES(?, ?, ?)"
+        ));
         sql::bytes id_bytes = uuid_get_bytes(data.get_id());
         statement->setBytes(1, &id_bytes);
-        std::optional<std::string> const& key = data.get_key();
-        if (key.has_value()) {
-            statement->setString(2, key.value());
-        } else {
-            statement->setNull(2, sql::DataType::VARCHAR);
-        }
-        statement->setString(3, data.get_value());
-        statement->setBoolean(4, data.is_hard_locality());
+        statement->setString(2, data.get_value());
+        statement->setBoolean(3, data.is_hard_locality());
         statement->executeUpdate();
 
         for (std::string const& addr : data.get_locality()) {
@@ -1219,6 +1227,13 @@ auto MySqlDataStorage::add_data(Data const& data) -> StorageErr {
             locality_statement->setString(2, addr);
             locality_statement->executeUpdate();
         }
+        std::unique_ptr<sql::PreparedStatement> driver_ref_statement(m_conn->prepareStatement(
+                "INSERT INTO `data_ref_driver` (`id`, `driver_id`) VALUES(?, ?)"
+        ));
+        sql::bytes driver_id_bytes = uuid_get_bytes(driver_id);
+        driver_ref_statement->setBytes(1, &id_bytes);
+        driver_ref_statement->setBytes(2, &driver_id_bytes);
+        driver_ref_statement->executeUpdate();
     } catch (sql::SQLException& e) {
         m_conn->rollback();
         if (e.getErrorCode() == ErDupKey || e.getErrorCode() == ErDupEntry) {
@@ -1230,74 +1245,68 @@ auto MySqlDataStorage::add_data(Data const& data) -> StorageErr {
     return StorageErr{};
 }
 
-auto MySqlDataStorage::get_data(boost::uuids::uuid id, Data* data) -> StorageErr {
+auto MySqlDataStorage::add_task_data(boost::uuids::uuid const task_id, Data const& data)
+        -> StorageErr {
     try {
-        std::unique_ptr<sql::PreparedStatement> statement(
-                m_conn->prepareStatement("SELECT `id`, `key`, `value`, `hard_locality` "
-                                         "FROM `data` WHERE `id` = ?")
-        );
-        sql::bytes id_bytes = uuid_get_bytes(id);
+        std::unique_ptr<sql::PreparedStatement> statement(m_conn->prepareStatement(
+                "INSERT INTO `data` (`id`, `value`, `hard_locality`) VALUES(?, ?, ?)"
+        ));
+        sql::bytes id_bytes = uuid_get_bytes(data.get_id());
         statement->setBytes(1, &id_bytes);
-        std::unique_ptr<sql::ResultSet> res(statement->executeQuery());
-        if (res->rowsCount() == 0) {
-            m_conn->rollback();
-            return StorageErr{
-                    StorageErrType::KeyNotFoundErr,
-                    fmt::format("no data with id {}", boost::uuids::to_string(id))
-            };
-        }
-        res->next();
-        if (res->isNull(2)) {
-            *data = Data{id, res->getString(3).c_str()};
-        } else {
-            *data = Data{id, res->getString(2).c_str(), res->getString(3).c_str()};
-        }
-        data->set_hard_locality(res->getBoolean(4));
+        statement->setString(2, data.get_value());
+        statement->setBoolean(3, data.is_hard_locality());
+        statement->executeUpdate();
 
-        std::unique_ptr<sql::PreparedStatement> locality_statement(
-                m_conn->prepareStatement("SELECT `address` FROM `data_locality` WHERE `id` = ?")
-        );
-        locality_statement->setBytes(1, &id_bytes);
-        std::unique_ptr<sql::ResultSet> const locality_res(locality_statement->executeQuery());
-        std::vector<std::string> locality;
-        while (locality_res->next()) {
-            locality.emplace_back(locality_res->getString(1));
-        }
-        if (!locality.empty()) {
-            data->set_locality(locality);
+        for (std::string const& addr : data.get_locality()) {
+            std::unique_ptr<sql::PreparedStatement> locality_statement(
+                    m_conn->prepareStatement("INSERT INTO `data_locality` (`id`, "
+                                             "`address`) VALUES (?, ?)")
+            );
+            locality_statement->setBytes(1, &id_bytes);
+            locality_statement->setString(2, addr);
+            locality_statement->executeUpdate();
         }
+        std::unique_ptr<sql::PreparedStatement> task_ref_statement(m_conn->prepareStatement(
+                "INSERT INTO `data_ref_task` (`id`, `task_id`) VALUES(?, ?)"
+        ));
+        sql::bytes task_id_bytes = uuid_get_bytes(task_id);
+        task_ref_statement->setBytes(1, &id_bytes);
+        task_ref_statement->setBytes(2, &task_id_bytes);
+        task_ref_statement->executeUpdate();
     } catch (sql::SQLException& e) {
         m_conn->rollback();
+        if (e.getErrorCode() == ErDupKey || e.getErrorCode() == ErDupEntry) {
+            return StorageErr{StorageErrType::DuplicateKeyErr, e.what()};
+        }
         return StorageErr{StorageErrType::OtherErr, e.what()};
     }
     m_conn->commit();
     return StorageErr{};
 }
 
-auto MySqlDataStorage::get_data_by_key(std::string const& key, Data* data) -> StorageErr {
+auto MySqlDataStorage::get_data(boost::uuids::uuid id, Data* data) -> StorageErr {
     try {
         std::unique_ptr<sql::PreparedStatement> statement(
-                m_conn->prepareStatement("SELECT `id`, `key`, `value`, `hard_locality` "
-                                         "FROM `data` WHERE `key` = ?")
+                m_conn->prepareStatement("SELECT `id`, `value`, `hard_locality` "
+                                         "FROM `data` WHERE `id` = ?")
         );
-        statement->setString(1, key);
+        sql::bytes id_bytes = uuid_get_bytes(id);
+        statement->setBytes(1, &id_bytes);
         std::unique_ptr<sql::ResultSet> res(statement->executeQuery());
         if (res->rowsCount() == 0) {
             m_conn->rollback();
             return StorageErr{
                     StorageErrType::KeyNotFoundErr,
-                    fmt::format("no data with key {}", key)
+                    fmt::format("no data with id {}", boost::uuids::to_string(id))
             };
         }
         res->next();
-        boost::uuids::uuid const id = read_id(res->getBinaryStream(1));
-        *data = Data{id, key, res->getString(3).c_str()};
-        data->set_hard_locality(res->getBoolean(4));
+        *data = Data{id, res->getString(2).c_str()};
+        data->set_hard_locality(res->getBoolean(3));
 
         std::unique_ptr<sql::PreparedStatement> locality_statement(
                 m_conn->prepareStatement("SELECT `address` FROM `data_locality` WHERE `id` = ?")
         );
-        sql::bytes id_bytes = uuid_get_bytes(id);
         locality_statement->setBytes(1, &id_bytes);
         std::unique_ptr<sql::ResultSet> const locality_res(locality_statement->executeQuery());
         std::vector<std::string> locality;
@@ -1416,4 +1425,131 @@ auto MySqlDataStorage::remove_driver_reference(boost::uuids::uuid id, boost::uui
     m_conn->commit();
     return StorageErr{};
 }
+
+auto MySqlDataStorage::remove_dangling_data() -> StorageErr {
+    try {
+        std::unique_ptr<sql::Statement> statement{m_conn->createStatement()};
+        statement->execute("DELETE FROM `data` WHERE `id` NOT IN (SELECT driver_ref.`id` FROM "
+                           "`data_ref_driver` driver_ref) AND `id` NOT IN (SELECT task_ref.`id` "
+                           "FROM `data_ref_task` task_ref)");
+    } catch (sql::SQLException& e) {
+        m_conn->rollback();
+        return StorageErr{StorageErrType::OtherErr, e.what()};
+    }
+    m_conn->commit();
+    return StorageErr{};
+}
+
+auto MySqlDataStorage::add_client_kv_data(KeyValueData const& data) -> StorageErr {
+    try {
+        std::unique_ptr<sql::PreparedStatement> statement(m_conn->prepareStatement(
+                "INSERT INTO `client_kv_data` (`kv_key`, `value`, `client_id`) VALUES(?, ?, ?)"
+        ));
+        statement->setString(1, data.get_key());
+        statement->setString(2, data.get_value());
+        sql::bytes id_bytes = uuid_get_bytes(data.get_id());
+        statement->setBytes(3, &id_bytes);
+        statement->executeUpdate();
+    } catch (sql::SQLException& e) {
+        m_conn->rollback();
+        if (e.getErrorCode() == ErDupKey || e.getErrorCode() == ErDupEntry) {
+            return StorageErr{StorageErrType::DuplicateKeyErr, e.what()};
+        }
+        return StorageErr{StorageErrType::OtherErr, e.what()};
+    }
+    m_conn->commit();
+    return StorageErr{};
+}
+
+auto MySqlDataStorage::add_task_kv_data(KeyValueData const& data) -> StorageErr {
+    try {
+        std::unique_ptr<sql::PreparedStatement> statement(m_conn->prepareStatement(
+                "INSERT INTO `task_kv_data` (`kv_key`, `value`, `task_id`) VALUES(?, ?, ?)"
+        ));
+        statement->setString(1, data.get_key());
+        statement->setString(2, data.get_value());
+        sql::bytes id_bytes = uuid_get_bytes(data.get_id());
+        statement->setBytes(3, &id_bytes);
+        statement->executeUpdate();
+    } catch (sql::SQLException& e) {
+        m_conn->rollback();
+        if (e.getErrorCode() == ErDupKey || e.getErrorCode() == ErDupEntry) {
+            return StorageErr{StorageErrType::DuplicateKeyErr, e.what()};
+        }
+        return StorageErr{StorageErrType::OtherErr, e.what()};
+    }
+    m_conn->commit();
+    return StorageErr{};
+}
+
+auto MySqlDataStorage::get_client_kv_data(
+        boost::uuids::uuid const& client_id,
+        std::string const& key,
+        std::string* value
+) -> StorageErr {
+    try {
+        std::unique_ptr<sql::PreparedStatement> statement(m_conn->prepareStatement(
+                "SELECT `value` "
+                "FROM `client_kv_data` WHERE `client_id` = ? AND `kv_key` = ?"
+        ));
+        sql::bytes id_bytes = uuid_get_bytes(client_id);
+        statement->setBytes(1, &id_bytes);
+        statement->setString(2, key);
+        std::unique_ptr<sql::ResultSet> res(statement->executeQuery());
+        if (res->rowsCount() == 0) {
+            m_conn->rollback();
+            return StorageErr{
+                    StorageErrType::KeyNotFoundErr,
+                    fmt::format(
+                            "no data for client {} with key {}",
+                            boost::uuids::to_string(client_id),
+                            key
+                    )
+            };
+        }
+        res->next();
+        *value = res->getString(1);
+    } catch (sql::SQLException& e) {
+        m_conn->rollback();
+        return StorageErr{StorageErrType::OtherErr, e.what()};
+    }
+    m_conn->commit();
+    return StorageErr{};
+}
+
+auto MySqlDataStorage::get_task_kv_data(
+        boost::uuids::uuid const& task_id,
+        std::string const& key,
+        std::string* value
+) -> StorageErr {
+    try {
+        std::unique_ptr<sql::PreparedStatement> statement(
+                m_conn->prepareStatement("SELECT `value` "
+                                         "FROM `task_kv_data` WHERE `task_id` = ? AND `kv_key` = ?")
+        );
+        sql::bytes id_bytes = uuid_get_bytes(task_id);
+        statement->setBytes(1, &id_bytes);
+        statement->setString(2, key);
+        std::unique_ptr<sql::ResultSet> res(statement->executeQuery());
+        if (res->rowsCount() == 0) {
+            m_conn->rollback();
+            return StorageErr{
+                    StorageErrType::KeyNotFoundErr,
+                    fmt::format(
+                            "no data for task {} with key {}",
+                            boost::uuids::to_string(task_id),
+                            key
+                    )
+            };
+        }
+        res->next();
+        *value = res->getString(1);
+    } catch (sql::SQLException& e) {
+        m_conn->rollback();
+        return StorageErr{StorageErrType::OtherErr, e.what()};
+    }
+    m_conn->commit();
+    return StorageErr{};
+}
+
 }  // namespace spider::core
diff --git a/src/spider/storage/MysqlStorage.hpp b/src/spider/storage/MysqlStorage.hpp
index 6a3938ce0..825c21608 100644
--- a/src/spider/storage/MysqlStorage.hpp
+++ b/src/spider/storage/MysqlStorage.hpp
@@ -13,6 +13,7 @@
 #include "../core/Data.hpp"
 #include "../core/Error.hpp"
 #include "../core/JobMetadata.hpp"
+#include "../core/KeyValueData.hpp"
 #include "../core/Task.hpp"
 #include "../core/TaskGraph.hpp"
 #include "DataStorage.hpp"
@@ -81,9 +82,9 @@ class MySqlDataStorage : public DataStorage {
     auto connect(std::string const& url) -> StorageErr override;
     void close() override;
     auto initialize() -> StorageErr override;
-    auto add_data(Data const& data) -> StorageErr override;
+    auto add_driver_data(boost::uuids::uuid driver_id, Data const& data) -> StorageErr override;
+    auto add_task_data(boost::uuids::uuid task_id, Data const& data) -> StorageErr override;
     auto get_data(boost::uuids::uuid id, Data* data) -> StorageErr override;
-    auto get_data_by_key(std::string const& key, Data* data) -> StorageErr override;
     auto remove_data(boost::uuids::uuid id) -> StorageErr override;
     auto
     add_task_reference(boost::uuids::uuid id, boost::uuids::uuid task_id) -> StorageErr override;
@@ -93,6 +94,20 @@ class MySqlDataStorage : public DataStorage {
             -> StorageErr override;
     auto remove_driver_reference(boost::uuids::uuid id, boost::uuids::uuid driver_id)
             -> StorageErr override;
+    auto remove_dangling_data() -> StorageErr override;
+
+    auto add_client_kv_data(KeyValueData const& data) -> StorageErr override;
+    auto add_task_kv_data(KeyValueData const& data) -> StorageErr override;
+    auto get_client_kv_data(
+            boost::uuids::uuid const& client_id,
+            std::string const& key,
+            std::string* value
+    ) -> StorageErr override;
+    auto get_task_kv_data(
+            boost::uuids::uuid const& task_id,
+            std::string const& key,
+            std::string* value
+    ) -> StorageErr override;
 
 private:
     sql::Connection* m_conn = nullptr;
diff --git a/src/spider/utils/TimedCache.hpp b/src/spider/utils/TimedCache.hpp
new file mode 100644
index 000000000..05b156039
--- /dev/null
+++ b/src/spider/utils/TimedCache.hpp
@@ -0,0 +1,54 @@
+#ifndef SPIDER_UTILS_TIMEDCACHE_HPP
+#define SPIDER_UTILS_TIMEDCACHE_HPP
+
+#include <chrono>
+#include <optional>
+#include <utility>
+
+#include <absl/container/flat_hash_map.h>
+
+namespace spider::core {
+
+constexpr unsigned cDefaultThreshold = 5;
+
+template <class Key, class Value>
+class TimedCache {
+public:
+    TimedCache() = default;
+
+    explicit TimedCache(unsigned const seconds) : m_duration{std::chrono::seconds(seconds)} {}
+
+    auto get(Key const& key) -> std::optional<Value> {
+        auto iter = m_map.find(key);
+        if (iter == m_map.end()) {
+            return std::nullopt;
+        }
+        iter->second.first = std::chrono::steady_clock::now();
+        return iter->second.second;
+    }
+
+    auto put(Key const& key, Value const& value) -> void {
+        auto iter = m_map.find(key);
+        if (iter == m_map.end()) {
+            m_map[key] = std::make_pair(std::chrono::steady_clock::now(), value);
+        } else {
+            iter->second.first = std::chrono::steady_clock::now();
+            iter->second.second = value;
+        }
+    }
+
+    auto cleanup() {
+        erase_if(m_map, [&](auto const& item) -> bool {
+            auto const& [key, value] = item;
+            return std::chrono::steady_clock::now() - value.first > m_duration;
+        });
+    }
+
+private:
+    std::chrono::steady_clock::duration m_duration = std::chrono::seconds(cDefaultThreshold);
+    absl::flat_hash_map<Key, std::pair<std::chrono::steady_clock::time_point, Value>> m_map;
+};
+
+}  // namespace spider::core
+
+#endif  // SPIDER_UTILS_TIMEDCACHE_HPP
diff --git a/tests/scheduler/test-SchedulerPolicy.cpp b/tests/scheduler/test-SchedulerPolicy.cpp
index eec1d84b7..16e2a0e26 100644
--- a/tests/scheduler/test-SchedulerPolicy.cpp
+++ b/tests/scheduler/test-SchedulerPolicy.cpp
@@ -83,16 +83,18 @@ TEMPLATE_LIST_TEST_CASE(
 
     boost::uuids::random_generator gen;
     boost::uuids::uuid const job_id = gen();
+    boost::uuids::uuid const client_id = gen();
     // Submit task with hard locality
     spider::core::Task task{"task"};
     spider::core::Data data{"value"};
     data.set_hard_locality(true);
     data.set_locality({"127.0.0.1"});
-    REQUIRE(data_store->add_data(data).success());
+    REQUIRE(metadata_store->add_driver(client_id, "127.0.0.1").success());
+    REQUIRE(data_store->add_driver_data(client_id, data).success());
     task.add_input(spider::core::TaskInput{data.get_id(), "int"});
     spider::core::TaskGraph graph;
     graph.add_task(task);
-    REQUIRE(metadata_store->add_job(job_id, gen(), graph).success());
+    REQUIRE(metadata_store->add_job(job_id, client_id, graph).success());
 
     spider::scheduler::FifoPolicy policy;
     // Schedule with wrong address
@@ -124,18 +126,20 @@ TEMPLATE_LIST_TEST_CASE(
             = std::move(std::get<0>(storages));
     std::shared_ptr<spider::core::DataStorage> const data_store = std::move(std::get<1>(storages));
 
+    // Add task
     boost::uuids::random_generator gen;
     boost::uuids::uuid const job_id = gen();
-    // Submit task with soft locality
+    boost::uuids::uuid const client_id = gen();
     spider::core::Task task{"task"};
     spider::core::Data data;
     data.set_hard_locality(false);
     data.set_locality({"127.0.0.1"});
-    REQUIRE(data_store->add_data(data).success());
+    REQUIRE(metadata_store->add_driver(client_id, "127.0.0.1").success());
+    REQUIRE(data_store->add_driver_data(client_id, data).success());
     task.add_input(spider::core::TaskInput{data.get_id(), "int"});
     spider::core::TaskGraph graph;
     graph.add_task(task);
-    REQUIRE(metadata_store->add_job(job_id, gen(), graph).success());
+    REQUIRE(metadata_store->add_job(job_id, client_id, graph).success());
 
     spider::scheduler::FifoPolicy policy;
     // Schedule with wrong address
diff --git a/tests/storage/test-DataStorage.cpp b/tests/storage/test-DataStorage.cpp
index 5841e66bc..e3a8ad4c3 100644
--- a/tests/storage/test-DataStorage.cpp
+++ b/tests/storage/test-DataStorage.cpp
@@ -1,5 +1,4 @@
 // NOLINTBEGIN(cert-err58-cpp,cppcoreguidelines-avoid-do-while,readability-function-cognitive-complexity,cppcoreguidelines-avoid-non-const-global-variables,cppcoreguidelines-avoid-c-arrays,modernize-avoid-c-arrays)
-#include <memory>
 #include <tuple>
 
 #include <boost/uuid/random_generator.hpp>
@@ -9,74 +8,108 @@
 
 #include "../../src/spider/core/Data.hpp"
 #include "../../src/spider/core/Error.hpp"
+#include "../../src/spider/core/KeyValueData.hpp"
 #include "../../src/spider/core/Task.hpp"
 #include "../../src/spider/core/TaskGraph.hpp"
-#include "../../src/spider/storage/DataStorage.hpp"
 #include "../utils/CoreDataUtils.hpp"
 #include "StorageTestHelper.hpp"
 
 namespace {
 
-TEMPLATE_LIST_TEST_CASE(
-        "Add, get and remove data",
-        "[storage]",
-        spider::test::DataStorageTypeList
-) {
-    std::unique_ptr<spider::core::DataStorage> storage
-            = spider::test::create_data_storage<TestType>();
+TEMPLATE_LIST_TEST_CASE("Add, get and remove data", "[storage]", spider::test::StorageTypeList) {
+    auto [metadata_storage, data_storage] = spider::test::
+            create_storage<std::tuple_element_t<0, TestType>, std::tuple_element_t<1, TestType>>();
 
-    // Add data
+    // Add driver and data
     spider::core::Data const data{"value"};
-    REQUIRE(storage->add_data(data).success());
+    boost::uuids::random_generator gen;
+    boost::uuids::uuid const driver_id = gen();
+    REQUIRE(metadata_storage->add_driver(driver_id, "127.0.0.1").success());
+    REQUIRE(data_storage->add_driver_data(driver_id, data).success());
 
     // Add data with same id again should fail
     spider::core::Data const data_same_id{data.get_id(), "value2"};
-    REQUIRE(spider::core::StorageErrType::DuplicateKeyErr == storage->add_data(data_same_id).type);
+    REQUIRE(spider::core::StorageErrType::DuplicateKeyErr
+            == data_storage->add_driver_data(driver_id, data_same_id).type);
 
     // Get data should match
     spider::core::Data result{"temp"};
-    REQUIRE(storage->get_data(data.get_id(), &result).success());
+    REQUIRE(data_storage->get_data(data.get_id(), &result).success());
     REQUIRE(spider::test::data_equal(data, result));
 
     // Remove data should succeed
-    REQUIRE(storage->remove_data(data.get_id()).success());
+    REQUIRE(data_storage->remove_data(data.get_id()).success());
 
     // Get data should fail
     REQUIRE(spider::core::StorageErrType::KeyNotFoundErr
-            == storage->get_data(data.get_id(), &result).type);
+            == data_storage->get_data(data.get_id(), &result).type);
 }
 
 TEMPLATE_LIST_TEST_CASE(
-        "Add, get and remove data with key",
+        "Add and get driver key value data",
         "[storage]",
-        spider::test::DataStorageTypeList
+        spider::test::StorageTypeList
 ) {
-    std::unique_ptr<spider::core::DataStorage> storage
-            = spider::test::create_data_storage<TestType>();
+    auto [metadata_storage, data_storage] = spider::test::
+            create_storage<std::tuple_element_t<0, TestType>, std::tuple_element_t<1, TestType>>();
+
+    // Add driver
+    boost::uuids::random_generator gen;
+    boost::uuids::uuid const driver_id = gen();
+    REQUIRE(metadata_storage->add_driver(driver_id, "127.0.0.1").success());
 
     // Add data
-    spider::core::Data const data{"key", "value"};
-    REQUIRE(storage->add_data(data).success());
+    spider::core::KeyValueData const data{"key", "value", driver_id};
+    REQUIRE(data_storage->add_client_kv_data(data).success());
 
-    // Add data with same key again should fail
-    spider::core::Data const data_same_key{"key", "value2"};
-    REQUIRE(spider::core::StorageErrType::DuplicateKeyErr == storage->add_data(data_same_key).type);
+    // Add data with same key and id again should fail
+    spider::core::KeyValueData const data_same_key{"key", "value2", driver_id};
+    REQUIRE(spider::core::StorageErrType::DuplicateKeyErr
+            == data_storage->add_client_kv_data(data_same_key).type);
 
     // Get data should match
-    spider::core::Data result{"temp"};
-    REQUIRE(storage->get_data_by_key("key", &result).success());
-    REQUIRE(spider::test::data_equal(data, result));
+    std::string value;
+    auto err = data_storage->get_client_kv_data(driver_id, "key", &value);
+    REQUIRE(data_storage->get_client_kv_data(driver_id, "key", &value).success());
+    REQUIRE(data.get_value() == value);
+}
 
-    // Remove data should succeed
-    REQUIRE(storage->remove_data(data.get_id()).success());
+TEMPLATE_LIST_TEST_CASE(
+        "Add and get task key value data",
+        "[storage]",
+        spider::test::StorageTypeList
+) {
+    auto [metadata_storage, data_storage] = spider::test::
+            create_storage<std::tuple_element_t<0, TestType>, std::tuple_element_t<1, TestType>>();
 
-    // Get data should fail
-    REQUIRE(spider::core::StorageErrType::KeyNotFoundErr
-            == storage->get_data_by_key("key", &result).type);
+    // Add task
+    boost::uuids::random_generator gen;
+    spider::core::Task const task{"func"};
+    spider::core::TaskGraph graph;
+    graph.add_task(task);
+    boost::uuids::uuid const job_id = gen();
+    REQUIRE(metadata_storage->add_job(job_id, gen(), graph).success());
+
+    // Add data
+    spider::core::KeyValueData const data{"key", "value", task.get_id()};
+    REQUIRE(data_storage->add_task_kv_data(data).success());
+
+    // Add data with same key and id again should fail
+    spider::core::KeyValueData const data_same_key{"key", "value2", task.get_id()};
+    REQUIRE(spider::core::StorageErrType::DuplicateKeyErr
+            == data_storage->add_task_kv_data(data_same_key).type);
+
+    // Get data should match
+    std::string value;
+    REQUIRE(data_storage->get_task_kv_data(task.get_id(), "key", &value).success());
+    REQUIRE(data.get_value() == value);
+
+    // Clean up
+    REQUIRE(metadata_storage->remove_job(job_id).success());
 }
 
 TEMPLATE_LIST_TEST_CASE(
-        "Add and remove data reference for task",
+        "Add and remove task reference for task",
         "[storage]",
         spider::test::StorageTypeList
 ) {
@@ -89,8 +122,10 @@ TEMPLATE_LIST_TEST_CASE(
 
     // Add task
     spider::core::Task const task{"func"};
+    spider::core::Task const task_2{"func"};
     spider::core::TaskGraph graph;
     graph.add_task(task);
+    graph.add_task(task_2);
     boost::uuids::uuid const job_id = gen();
     REQUIRE(metadata_storage->add_job(job_id, gen(), graph).success());
 
@@ -99,16 +134,24 @@ TEMPLATE_LIST_TEST_CASE(
 
     // Add data
     spider::core::Data const data{"value"};
-    REQUIRE(data_storage->add_data(data).success());
+    REQUIRE(data_storage->add_task_data(task.get_id(), data).success());
 
     // Add task reference
-    REQUIRE(data_storage->add_task_reference(data.get_id(), task.get_id()).success());
+    REQUIRE(data_storage->add_task_reference(data.get_id(), task_2.get_id()).success());
 
     // Remove task reference
-    REQUIRE(data_storage->remove_task_reference(data.get_id(), task.get_id()).success());
+    REQUIRE(data_storage->remove_task_reference(data.get_id(), task_2.get_id()).success());
 
     // Remove job
     REQUIRE(metadata_storage->remove_job(job_id).success());
+
+    // Clean up
+    REQUIRE(data_storage->remove_dangling_data().success());
+
+    // Get data should fail
+    spider::core::Data res{"temp"};
+    REQUIRE(spider::core::StorageErrType::KeyNotFoundErr
+            == data_storage->get_data(data.get_id(), &res).type);
 }
 
 TEMPLATE_LIST_TEST_CASE(
@@ -126,20 +169,22 @@ TEMPLATE_LIST_TEST_CASE(
 
     // Add driver
     boost::uuids::uuid const driver_id = gen();
+    boost::uuids::uuid const driver_id_2 = gen();
     REQUIRE(metadata_storage->add_driver(driver_id, "127.0.0.1").success());
+    REQUIRE(metadata_storage->add_driver(driver_id_2, "127.0.0.1").success());
 
     // Add driver reference without data should fail
     REQUIRE(!data_storage->add_driver_reference(gen(), driver_id).success());
 
     // Add data
     spider::core::Data const data{"value"};
-    REQUIRE(data_storage->add_data(data).success());
+    REQUIRE(data_storage->add_driver_data(driver_id, data).success());
 
     // Add driver reference
-    REQUIRE(data_storage->add_driver_reference(data.get_id(), driver_id).success());
+    REQUIRE(data_storage->add_driver_reference(data.get_id(), driver_id_2).success());
 
     // Remove driver reference
-    REQUIRE(data_storage->remove_driver_reference(data.get_id(), driver_id).success());
+    REQUIRE(data_storage->remove_driver_reference(data.get_id(), driver_id_2).success());
 }
 }  // namespace
 
diff --git a/tests/utils/CoreDataUtils.hpp b/tests/utils/CoreDataUtils.hpp
index 99eebeede..8354db8be 100644
--- a/tests/utils/CoreDataUtils.hpp
+++ b/tests/utils/CoreDataUtils.hpp
@@ -9,10 +9,6 @@ inline auto data_equal(core::Data const& d1, core::Data const& d2) -> bool {
         return false;
     }
 
-    if (d1.get_key() != d2.get_key()) {
-        return false;
-    }
-
     if (d1.get_locality() != d2.get_locality()) {
         return false;
     }
diff --git a/tests/worker/test-FunctionManager.cpp b/tests/worker/test-FunctionManager.cpp
index a1b109be9..5df5db360 100644
--- a/tests/worker/test-FunctionManager.cpp
+++ b/tests/worker/test-FunctionManager.cpp
@@ -5,7 +5,6 @@
 
 #include <catch2/catch_test_macros.hpp>
 
-#include "../../src/spider/core/Data.hpp"
 #include "../../src/spider/io/MsgPack.hpp"  // IWYU pragma: keep
 #include "../../src/spider/worker/FunctionManager.hpp"
 
@@ -18,13 +17,8 @@ auto tuple_ret_test(std::string const& str, int const x) -> std::tuple<std::stri
     return std::make_tuple(str, x);
 }
 
-auto data_test(spider::core::Data const& data) -> spider::core::Data {
-    return spider::core::Data{data.get_id(), data.get_value() + data.get_value()};
-}
-
 SPIDER_WORKER_REGISTER_TASK(int_test);
 SPIDER_WORKER_REGISTER_TASK(tuple_ret_test);
-SPIDER_WORKER_REGISTER_TASK(data_test);
 
 TEST_CASE("Register and run function with POD inputs", "[core]") {
     spider::core::FunctionManager const& manager = spider::core::FunctionManager::get_instance();
@@ -75,24 +69,6 @@ TEST_CASE("Register and run function with tuple return", "[core]") {
             ));
 }
 
-TEST_CASE("Register and run function with data", "[core]") {
-    spider::core::FunctionManager const& manager = spider::core::FunctionManager::get_instance();
-
-    spider::core::Function const* function = manager.get_function("data_test");
-
-    spider::core::Data data{"test"};
-    spider::core::ArgsBuffer const args_buffers = spider::core::create_args_buffers(data);
-    msgpack::sbuffer const result = (*function)(args_buffers);
-    std::optional<spider::core::Data> result_option
-            = spider::core::response_get_result<spider::core::Data>(result);
-    REQUIRE(result_option.has_value());
-    if (result_option.has_value()) {
-        spider::core::Data const& result_data = result_option.value();
-        REQUIRE(data.get_id() == result_data.get_id());
-        REQUIRE("testtest" == result_data.get_value());
-    }
-}
-
 }  // namespace
 
 // NOLINTEND(cert-err58-cpp,cppcoreguidelines-avoid-do-while,readability-function-cognitive-complexity,cppcoreguidelines-avoid-non-const-global-variables,cppcoreguidelines-avoid-c-arrays,modernize-avoid-c-arrays)