Fix invalid storage dir configurations lead to unexpected behavior (#4105)

dbms/src/Common/tests/gtest_cpptoml.cpp

@@ -0,0 +1,149 @@
+#include <TestUtils/TiFlashTestBasic.h>
+#include <common/logger_useful.h>
+/// Suppress gcc warning: ‘*((void*)&<anonymous> +4)’ may be used uninitialized in this function
+#if !__clang__
+#pragma GCC diagnostic push
+#pragma GCC diagnostic ignored "-Wmaybe-uninitialized"
+#endif
+#include <Common/Config/cpptoml.h>
+#if !__clang__
+#pragma GCC diagnostic pop
+#endif
+
+#include <sstream>
+
+namespace DB::tests
+{
+
+TEST(CPPTomlTest, ContainsQualifiedArray)
+{
+    auto * log = &Poco::Logger::get("CPPTomlTest");
+
+    Strings failure_tests = {
+        R"(
+[a]
+[a.b]
+c = "abc"
+        )",
+        R"(
+[a]
+[a.b]
+c = 123
+        )",
+        R"(
+[a]
+[a.b]
+c = 123.45
+        )",
+    };
+
+    for (size_t i = 0; i < failure_tests.size(); ++i)
+    {
+        const auto & test_case = failure_tests[i];
+        SCOPED_TRACE(fmt::format("[index={}] [content={}]", i, test_case));
+        LOG_INFO(log, "parsing [index=" << i << "] [content=" << test_case << "]");
+
+        std::istringstream ss(test_case);
+        cpptoml::parser p(ss);
+        auto table = p.parse();
+
+        const char * key = "a.b.c";
+        ASSERT_TRUE(table->contains_qualified(key));
+        auto qualified = table->get_qualified(key);
+        ASSERT_TRUE(qualified);
+        // not array
+        ASSERT_FALSE(qualified->is_array());
+        // try to parse as vector<string>, return false
+        cpptoml::option<Strings> array = table->get_qualified_array_of<String>(key);
+        ASSERT_FALSE(array);
+    }
+}
+
+TEST(CPPTomlTest, ContainsQualifiedStringArray)
+{
+    auto * log = &Poco::Logger::get("CPPTomlTest");
+
+    Strings failure_tests = {
+        R"(
+[a]
+[a.b]
+c = [["abc", "def"], ["z"]]
+        )",
+        R"(
+[a]
+[a.b]
+c = [123, 456]
+        )",
+    };
+
+    for (size_t i = 0; i < failure_tests.size(); ++i)
+    {
+        const auto & test_case = failure_tests[i];
+        SCOPED_TRACE(fmt::format("[index={}] [content={}]", i, test_case));
+        LOG_INFO(log, "parsing [index=" << i << "] [content=" << test_case << "]");
+
+        std::istringstream ss(test_case);
+        cpptoml::parser p(ss);
+        auto table = p.parse();
+
+        const char * key = "a.b.c";
+        ASSERT_TRUE(table->contains_qualified(key));
+        auto qualified = table->get_qualified(key);
+        ASSERT_TRUE(qualified);
+        // is non-empty array but not string array
+        ASSERT_TRUE(qualified->is_array());
+        auto qualified_array = qualified->as_array();
+        ASSERT_NE(qualified_array->begin(), qualified_array->end());
+        // try to parse as vector<string>, return false
+        cpptoml::option<Strings> array = table->get_qualified_array_of<String>(key);
+        ASSERT_FALSE(array);
+    }
+}
+
+TEST(CPPTomlTest, ContainsQualifiedStringArrayOrEmpty)
+{
+    auto * log = &Poco::Logger::get("CPPTomlTest");
+
+    Strings failure_tests = {
+        // a.b.c is not empty
+        R"(
+[a]
+[a.b]
+c = ["abc", "def", "z"]
+        )",
+        // a.b.c is empty
+        R"(
+[a]
+[a.b]
+c = []
+        )",
+    };
+
+    for (size_t i = 0; i < failure_tests.size(); ++i)
+    {
+        const auto & test_case = failure_tests[i];
+        SCOPED_TRACE(fmt::format("[index={}] [content={}]", i, test_case));
+        LOG_INFO(log, "parsing [index=" << i << "] [content=" << test_case << "]");
+
+        std::istringstream ss(test_case);
+        cpptoml::parser p(ss);
+        auto table = p.parse();
+
+        const char * key = "a.b.c";
+        ASSERT_TRUE(table->contains_qualified(key));
+        auto qualified = table->get_qualified(key);
+        ASSERT_TRUE(qualified);
+        // is non-empty array but not string array
+        ASSERT_TRUE(qualified->is_array());
+
+        // try to parse as vector<string>, return true
+        cpptoml::option<Strings> array = table->get_qualified_array_of<String>(key);
+        ASSERT_TRUE(array);
+        if (auto qualified_array = qualified->as_array(); qualified_array->begin() != qualified_array->end())
+        {
+            ASSERT_EQ(array->size(), 3);
+        }
+    }
+}
+
+} // namespace DB::tests

dbms/src/Server/StorageConfigParser.cpp

@@ -50,8 +50,34 @@ void TiFlashStorageConfig::parseStoragePath(const String & storage, Poco::Logger
     cpptoml::parser p(ss);
     auto table = p.parse();
 
+    auto get_checked_qualified_array = [log](const std::shared_ptr<cpptoml::table> table, const char * key) -> cpptoml::option<Strings> {
+        auto throw_invalid_value = [log, key]() {
+            String error_msg
+                = fmt::format("The configuration \"storage.{}\" should be an array of strings. Please check your configuration file.", key);
+            LOG_ERROR(log, error_msg);
+            throw Exception(error_msg, ErrorCodes::INVALID_CONFIG_PARAMETER);
+        };
+        // not exist key
+        if (!table->contains_qualified(key))
+            return cpptoml::option<Strings>();
+
+        // key exist, but not array
+        auto qualified_ptr = table->get_qualified(key);
+        if (!qualified_ptr->is_array())
+        {
+            throw_invalid_value();
+        }
+        // key exist, but can not convert to string array, maybe it is an int array
+        auto string_array = table->get_qualified_array_of<String>(key);
+        if (!string_array)
+        {
+            throw_invalid_value();
+        }
+        return string_array;
+    };
+
     // main
-    if (auto main_paths = table->get_qualified_array_of<String>("main.dir"); main_paths)
+    if (auto main_paths = get_checked_qualified_array(table, "main.dir"); main_paths)
         main_data_paths = *main_paths;
     if (auto main_capacity = table->get_qualified_array_of<int64_t>("main.capacity"); main_capacity)
     {
@@ -66,9 +92,11 @@ void TiFlashStorageConfig::parseStoragePath(const String & storage, Poco::Logger
     }
     if (!main_capacity_quota.empty() && main_capacity_quota.size() != main_data_paths.size())
     {
-        String error_msg = "The array size of \"storage.main.dir\"[size=" + toString(main_data_paths.size())
-            + "] is not equal to \"storage.main.capacity\"[size=" + toString(main_capacity_quota.size())
-            + "]. Please check your configuration file.";
+        String error_msg = fmt::format("The array size of \"storage.main.dir\"[size={}] "
+                                       "is not equal to \"storage.main.capacity\"[size={}]. "
+                                       "Please check your configuration file.",
+            main_data_paths.size(),
+            main_capacity_quota.size());
         LOG_ERROR(log, error_msg);
         throw Exception(error_msg, ErrorCodes::INVALID_CONFIG_PARAMETER);
     }
@@ -82,13 +110,14 @@ void TiFlashStorageConfig::parseStoragePath(const String & storage, Poco::Logger
     }
 
     // latest
-    if (auto latest_paths = table->get_qualified_array_of<String>("latest.dir"); latest_paths)
+    if (auto latest_paths = get_checked_qualified_array(table, "latest.dir"); latest_paths)
         latest_data_paths = *latest_paths;
     if (auto latest_capacity = table->get_qualified_array_of<int64_t>("latest.capacity"); latest_capacity)
     {
         for (const auto & c : *latest_capacity)
             latest_capacity_quota.emplace_back((size_t)c);
     }
+    // If it is empty, use the same dir as "main.dir"
     if (latest_data_paths.empty())
     {
         LOG_INFO(log, "The configuration \"storage.latest.dir\" is empty, use the same dir and capacity of \"storage.main.dir\"");
@@ -97,9 +126,11 @@ void TiFlashStorageConfig::parseStoragePath(const String & storage, Poco::Logger
     }
     if (!latest_capacity_quota.empty() && latest_capacity_quota.size() != latest_data_paths.size())
     {
-        String error_msg = "The array size of \"storage.main.dir\"[size=" + toString(latest_data_paths.size())
-            + "] is not euqal to \"storage.main.capacity\"[size=" + toString(latest_capacity_quota.size())
-            + "]. Please check your configuration file.";
+        String error_msg = fmt::format("The array size of \"storage.latest.dir\"[size={}] "
+                                       "is not equal to \"storage.latest.capacity\"[size={}]. "
+                                       "Please check your configuration file.",
+            latest_data_paths.size(),
+            latest_capacity_quota.size());
         LOG_ERROR(log, error_msg);
         throw Exception(error_msg, ErrorCodes::INVALID_CONFIG_PARAMETER);
     }
@@ -113,7 +144,7 @@ void TiFlashStorageConfig::parseStoragePath(const String & storage, Poco::Logger
     }
 
     // Raft
-    if (auto kvstore_paths = table->get_qualified_array_of<String>("raft.dir"); kvstore_paths)
+    if (auto kvstore_paths = get_checked_qualified_array(table, "raft.dir"); kvstore_paths)
         kvstore_data_path = *kvstore_paths;
     if (kvstore_data_path.empty())
     {
@@ -307,7 +338,7 @@ std::tuple<size_t, TiFlashStorageConfig> TiFlashStorageConfig::parseSettings(Poc
         if (!storage_config.parseFromDeprecatedConfiguration(config, log))
         {
             // Can not parse from the deprecated configuration "path".
-            String msg = "The configuration \"storage\" section is not defined. Please check your configuration file.";
+            String msg = "The configuration \"storage.main\" section is not defined. Please check your configuration file.";
             LOG_ERROR(log, msg);
             throw Exception(msg, ErrorCodes::INVALID_CONFIG_PARAMETER);
         }
@@ -335,7 +366,7 @@ void StorageIORateLimitConfig::parse(const String & storage_io_rate_limit, Poco:
     readConfig("foreground_write_weight", fg_write_weight);
     readConfig("background_write_weight", bg_write_weight);
     readConfig("foreground_read_weight", fg_read_weight);
-    readConfig("background_read_weight", bg_read_weight);    
+    readConfig("background_read_weight", bg_read_weight);
     readConfig("emergency_pct", emergency_pct);
     readConfig("high_pct", high_pct);
     readConfig("medium_pct", medium_pct);