WALStore support multi path

dbms/src/Storages/Page/V3/WAL/WALReader.cpp

@@ -22,17 +22,21 @@ LogFilenameSet WALStoreReader::listAllFiles(
     // [<parent_path_0, [file0, file1, ...]>, <parent_path_1, [...]>, ...]
     std::vector<std::pair<String, Strings>> all_filenames;
     Strings filenames;
-    for (const auto & p : delegator->listPaths())
+    for (const auto & parent_path : delegator->listPaths())
     {
-        Poco::File directory(p);
+        String wal_parent_path = parent_path + WALStore::wal_folder_prefix;
+        Poco::File directory(wal_parent_path);
         if (!directory.exists())
+        {
             directory.createDirectories();
+            continue;
+        }
+
         filenames.clear();
         directory.list(filenames);
-        all_filenames.emplace_back(std::make_pair(p, std::move(filenames)));
+        all_filenames.emplace_back(std::make_pair(wal_parent_path, std::move(filenames)));
         filenames.clear();
     }
-    assert(all_filenames.size() == 1); // TODO: multi-path
 
     LogFilenameSet log_files;
     for (const auto & [parent_path, filenames] : all_filenames)

dbms/src/Storages/Page/V3/WALStore.cpp

@@ -105,6 +105,8 @@ void WALStore::apply(const PageEntriesEdit & edit)
     }
 }
 
+UInt16 WALStore::wal_paths_index = 0;
+
 std::tuple<std::unique_ptr<LogWriter>, LogFilename> WALStore::createLogWriter(
     PSDiskDelegatorPtr delegator,
     const FileProviderPtr & provider,
@@ -113,7 +115,26 @@ std::tuple<std::unique_ptr<LogWriter>, LogFilename> WALStore::createLogWriter(
     Poco::Logger * logger,
     bool manual_flush)
 {
-    const auto path = delegator->defaultPath(); // TODO: multi-path
+    String path;
+
+    if (delegator->numPaths() == 1)
+    {
+        path = delegator->defaultPath();
+    }
+    else
+    {
+        const auto & paths = delegator->listPaths();
+
+        if (wal_paths_index >= paths.size())
+        {
+            wal_paths_index = 0;
+        }
+        path = paths[wal_paths_index];
+        wal_paths_index++;
+    }
+
+    path += wal_folder_prefix;
+
     LogFilename log_filename = LogFilename{
         (manual_flush ? LogFileStage::Temporary : LogFileStage::Normal),
         new_log_lvl.first,

dbms/src/Storages/Page/V3/WALStore.h

@@ -68,6 +68,7 @@ using WALStoreReaderPtr = std::shared_ptr<WALStoreReader>;
 class WALStore
 {
 public:
+    constexpr static const char * wal_folder_prefix = "/wal";
     using ChecksumClass = Digest::CRC64;
 
     static WALStorePtr create(
@@ -104,6 +105,8 @@ class WALStore
     std::unique_ptr<LogWriter> log_file;
 
     Poco::Logger * logger;
+
+    static UInt16 wal_paths_index;
 };
 
 } // namespace PS::V3