feat: support partition group

include/async_io_manager.h

@@ -355,6 +355,7 @@ class AsyncIoManager
 };
 
 KvError ToKvError(int err_no);
+int KvErrorToIoResult(KvError err);
 
 class IouringMgr : public AsyncIoManager
 {

include/eloq_store.h

@@ -10,6 +10,7 @@
 #include <limits>
 #include <memory>
 #include <mutex>
+#include <optional>
 #include <string>
 #include <vector>
 
@@ -51,7 +52,9 @@ enum class RequestType : uint8_t
     CleanExpired,
     GlobalArchive,
     GlobalReopen,
-    GlobalListArchiveTags
+    GlobalListArchiveTags,
+    ChangePartition,
+    ApplyPartitionChange
 };
 
 inline const char *RequestTypeToString(RequestType type)
@@ -90,6 +93,10 @@ inline const char *RequestTypeToString(RequestType type)
         return "global_reopen";
     case RequestType::GlobalListArchiveTags:
         return "global_list_archive_tags";
+    case RequestType::ChangePartition:
+        return "change_partition";
+    case RequestType::ApplyPartitionChange:
+        return "apply_partition_change";
     default:
         return "unknown";
     }
@@ -617,6 +624,55 @@ class GlobalListArchiveTagsRequest : public KvRequest
     friend class EloqStore;
 };
 
+class ChangePartitionRequest : public KvRequest
+{
+public:
+    RequestType Type() const override
+    {
+        return RequestType::ChangePartition;
+    }
+
+    void SetPartitionFilter(PartitionFilter partition_filter)
+    {
+        if (partition_filter)
+        {
+            partition_filter_ = std::make_shared<PartitionFilter>(
+                std::move(partition_filter));
+        }
+        else
+        {
+            partition_filter_.reset();
+        }
+    }
+
+    void SetStopStore(bool stop_store)
+    {
+        stop_store_ = stop_store;
+    }
+
+private:
+    std::shared_ptr<const PartitionFilter> partition_filter_{nullptr};
+    bool stop_store_{false};
+
+    friend class EloqStore;
+};
+
+class ApplyPartitionChangeRequest : public KvRequest
+{
+public:
+    RequestType Type() const override
+    {
+        return RequestType::ApplyPartitionChange;
+    }
+
+private:
+    std::shared_ptr<const PartitionFilter> partition_filter_{nullptr};
+    std::atomic<uint32_t> *pending_{nullptr};
+
+    friend class EloqStore;
+    friend class Shard;
+};
+
 class CompactRequest : public WriteRequest
 {
 public:
@@ -747,6 +803,10 @@ class EloqStore
     void HandleGlobalArchiveRequest(GlobalArchiveRequest *req);
     void HandleGlobalReopenRequest(GlobalReopenRequest *req);
     void HandleGlobalListArchiveTagsRequest(GlobalListArchiveTagsRequest *req);
+    void HandleChangePartitionRequest(ChangePartitionRequest *req);
+    std::optional<PartitionGroupId> ResolvePartitionGroup(
+        const TableIdent &tbl_id) const;
+    bool PartitionIncluded(const TableIdent &tbl_id) const;
     KvError CollectTablePartitions(const std::string &table_name,
                                    std::vector<TableIdent> &partitions) const;
     KvError InitStoreSpace();
@@ -773,6 +833,7 @@ class EloqStore
 
     bool enable_eloqstore_metrics_{false};
     std::atomic<StoreMode> store_mode_{StoreMode::Local};
+    std::shared_ptr<const PartitionFilter> partition_filter_{nullptr};
 
     friend class Shard;
     friend class AsyncIoManager;

include/error.h

@@ -27,6 +27,7 @@ enum struct KvError : uint8_t
     NoPermission,   // Permission denied (EPERM).
     CloudErr,       // Cloud service error (non-timeout HTTP/CURL).
     IoFail,         // Unclassified local I/O error.
+    Aborted,        // Operation aborted after in-flight I/O drained.
     ExpiredTerm,    // Cloud term file indicates stale process term.
     OssInsufficientStorage,  // Object storage out of capacity (HTTP 507).
 };
@@ -59,6 +60,8 @@ constexpr const char *ErrorString(KvError err)
         return "Device or resource busy";
     case KvError::IoFail:
         return "I/O failure";
+    case KvError::Aborted:
+        return "Operation aborted";
     case KvError::CloudErr:
         return "Cloud service is unavailable";
     case KvError::Timeout:

include/kv_options.h

@@ -5,6 +5,7 @@
 #include <cstdint>
 #include <cstring>
 #include <functional>
+#include <optional>
 #include <string>
 #include <vector>
 
@@ -20,6 +21,9 @@ constexpr int64_t TB = 1LL << 40;
 
 constexpr uint8_t max_overflow_pointers = 128;
 constexpr uint16_t max_read_pages_batch = max_overflow_pointers;
+using PartitionFilter =
+    std::function<std::optional<PartitionGroupId>(const TableIdent &)>;
+
 struct KvOptions
 {
     int LoadFromIni(const char *path);
@@ -304,15 +308,15 @@ struct KvOptions
     uint16_t prewarm_task_count = 3;
 
     /**
-     * @brief Filter function to determine which partitions belong to this
-     * instance.
-     * The filter returning true means that this table partition belongs to the
-     * current instance and should be included in operations like prewarming,
-     * snapshotting, etc.
+     * @brief Classify which partition group a table partition belongs to.
+     * Returning std::nullopt means that this partition does not belong to this
+     * instance. If the function returns a partition group id, runtime
+     * partition-group state decides whether that partition is currently active
+     * for operations like prewarming, snapshotting, and reopen.
      * If not set (empty), all partitions are considered to belong to this
      * instance.
      */
-    std::function<bool(const TableIdent &)> partition_filter;
+    PartitionFilter partition_filter;
 };
 
 }  // namespace eloqstore

include/storage/shard.h

@@ -77,6 +77,22 @@ class Shard
     void OnTaskFinished(KvTask *task);
     void OnReceivedReq(KvRequest *req);
     bool ProcessReq(KvRequest *req);
+    static bool ShouldCheckPartitionFilter(RequestType type)
+    {
+        switch (type)
+        {
+        case RequestType::ListObject:
+        case RequestType::ListStandbyPartition:
+        case RequestType::GlobalArchive:
+        case RequestType::GlobalReopen:
+        case RequestType::GlobalListArchiveTags:
+        case RequestType::ChangePartition:
+        case RequestType::ApplyPartitionChange:
+            return false;
+        default:
+            return true;
+        }
+    }
     void TryStartPendingWrite(const TableIdent &tbl_id);
     void TryDispatchPendingWrites();
 

include/tasks/task.h

@@ -115,10 +115,19 @@ class KvTask
     int WaitIoResult();
     void WaitIo();
     void FinishIo();
+    void ForceAbort()
+    {
+        force_aborted_ = true;
+    }
+    bool ForceAborted() const
+    {
+        return force_aborted_;
+    }
 
     uint32_t inflight_io_{0};
     int io_res_{0};
     uint32_t io_flags_{0};
+    bool force_aborted_{false};
 
     TaskStatus status_{TaskStatus::Idle};
     KvRequest *req_{nullptr};

include/tasks/task_manager.h

@@ -44,6 +44,26 @@ class TaskManager
     void FinishExternalTask();
 
     size_t NumActive() const;
+    template <typename F>
+    void ForEachActiveTask(F &&visitor)
+    {
+        auto visit = [&](KvTask *task)
+        {
+            if (task->status_ == TaskStatus::Idle ||
+                task->status_ == TaskStatus::Finished)
+            {
+                return;
+            }
+            visitor(task);
+        };
+        batch_write_pool_.ForEachTask(visit);
+        bg_write_pool_.ForEachTask(visit);
+        read_pool_.ForEachTask(visit);
+        scan_pool_.ForEachTask(visit);
+        list_object_pool_.ForEachTask(visit);
+        list_standby_partition_pool_.ForEachTask(visit);
+        reopen_pool_.ForEachTask(visit);
+    }
 
     void Shutdown();
 

include/types.h

@@ -14,6 +14,8 @@
 
 namespace eloqstore
 {
+using PartitionGroupId = uint32_t;
+
 enum class StoreMode
 {
     Local = 0,

src/async_io_manager.cpp

@@ -992,12 +992,41 @@ KvError ToKvError(int err_no)
         return KvError::OpenFileLimit;
     case -ENOSPC:
         return KvError::OutOfSpace;
+    case -ECANCELED:
+        return KvError::Aborted;
     default:
         LOG(ERROR) << "ToKvError: " << err_no;
         return KvError::IoFail;
     }
 }
 
+int KvErrorToIoResult(KvError err)
+{
+    switch (err)
+    {
+    case KvError::NoError:
+        return 0;
+    case KvError::NoPermission:
+        return -EPERM;
+    case KvError::NotFound:
+        return -ENOENT;
+    case KvError::TryAgain:
+        return -EAGAIN;
+    case KvError::OutOfMem:
+        return -ENOMEM;
+    case KvError::Busy:
+        return -EBUSY;
+    case KvError::OpenFileLimit:
+        return -EMFILE;
+    case KvError::OutOfSpace:
+        return -ENOSPC;
+    case KvError::Aborted:
+        return -ECANCELED;
+    default:
+        return -EIO;
+    }
+}
+
 std::pair<void *, IouringMgr::UserDataType> IouringMgr::DecodeUserData(
     uint64_t user_data)
 {
@@ -4673,7 +4702,7 @@ std::pair<ManifestFilePtr, KvError> StandbyStoreMgr::GetManifest(
         return {nullptr, prep_err};
     }
     current_task->WaitIo();
-    prep_err = static_cast<KvError>(current_task->io_res_);
+    prep_err = ToKvError(current_task->io_res_);
     if (prep_err != KvError::NoError)
     {
         return {nullptr, prep_err};