Merge pull request #43584 from ClickHouse/fix_config_child_list_iteration

Iterate list without index-based access

src/Common/Config/ConfigProcessor.cpp

@@ -96,9 +96,8 @@ static ElementIdentifier getElementIdentifier(Node * element)
 {
     const NamedNodeMapPtr attrs = element->attributes();
     std::vector<std::pair<std::string, std::string>> attrs_kv;
-    for (size_t i = 0, size = attrs->length(); i < size; ++i)
+    for (const Node * node = attrs->item(0); node; node = node->nextSibling())
     {
-        const Node * node = attrs->item(i);
         std::string name = node->nodeName();
         const auto * subst_name_pos = std::find(ConfigProcessor::SUBSTITUTION_ATTRS.begin(), ConfigProcessor::SUBSTITUTION_ATTRS.end(), name);
         if (name == "replace" || name == "remove" ||
@@ -123,9 +122,8 @@ static ElementIdentifier getElementIdentifier(Node * element)
 static Node * getRootNode(Document * document)
 {
     const NodeListPtr children = document->childNodes();
-    for (size_t i = 0, size = children->length(); i < size; ++i)
+    for (Node * child = children->item(0); child; child = child->nextSibling())
     {
-        Node * child = children->item(i);
         /// Besides the root element there can be comment nodes on the top level.
         /// Skip them.
         if (child->nodeType() == Node::ELEMENT_NODE)
@@ -145,10 +143,8 @@ static void deleteAttributesRecursive(Node * root)
     const NodeListPtr children = root->childNodes();
     std::vector<Node *> children_to_delete;
 
-    for (size_t i = 0, size = children->length(); i < size; ++i)
+    for (Node * child = children->item(0); child; child = child->nextSibling())
     {
-        Node * child = children->item(i);
-
         if (child->nodeType() == Node::ELEMENT_NODE)
         {
             Element & child_element = dynamic_cast<Element &>(*child);
@@ -189,10 +185,10 @@ void ConfigProcessor::mergeRecursive(XMLDocumentPtr config, Node * config_root,
         node = next_node;
     }
 
-    for (size_t i = 0, size = with_nodes->length(); i < size; ++i)
+    Node * next_with_node = nullptr;
+    for (Node * with_node = with_nodes->item(0); with_node; with_node = next_with_node)
     {
-        Node * with_node = with_nodes->item(i);
-
+        next_with_node = with_node->nextSibling();
         bool merged = false;
         bool remove = false;
         if (with_node->nodeType() == Node::ELEMENT_NODE)
@@ -342,9 +338,11 @@ void ConfigProcessor::doIncludesRecursive(
             if (node->nodeName() == "include")
             {
                 const NodeListPtr children = node_to_include->childNodes();
-                for (size_t i = 0, size = children->length(); i < size; ++i)
+                Node * next_child = nullptr;
+                for (Node * child = children->item(0); child; child = next_child)
                 {
-                    NodePtr new_node = config->importNode(children->item(i), true);
+                    next_child = child->nextSibling();
+                    NodePtr new_node = config->importNode(child, true);
                     node->parentNode()->insertBefore(new_node, node);
                 }
 
@@ -366,16 +364,20 @@ void ConfigProcessor::doIncludesRecursive(
                 }
 
                 const NodeListPtr children = node_to_include->childNodes();
-                for (size_t i = 0, size = children->length(); i < size; ++i)
+                Node * next_child = nullptr;
+                for (Node * child = children->item(0); child; child = next_child)
                 {
-                    NodePtr new_node = config->importNode(children->item(i), true);
+                    next_child = child->nextSibling();
+                    NodePtr new_node = config->importNode(child, true);
                     node->appendChild(new_node);
                 }
 
                 const NamedNodeMapPtr from_attrs = node_to_include->attributes();
-                for (size_t i = 0, size = from_attrs->length(); i < size; ++i)
+                Node * next_attr = nullptr;
+                for (Node * attr = from_attrs->item(0); attr; attr = next_attr)
                 {
-                    element.setAttributeNode(dynamic_cast<Attr *>(config->importNode(from_attrs->item(i), true)));
+                    next_attr = attr->nextSibling();
+                    element.setAttributeNode(dynamic_cast<Attr *>(config->importNode(attr, true)));
                 }
 
                 included_something = true;
@@ -437,9 +439,12 @@ void ConfigProcessor::doIncludesRecursive(
     else
     {
         NodeListPtr children = node->childNodes();
-        Node * child = nullptr;
-        for (size_t i = 0; (child = children->item(i)); ++i)
+        Node * next_child = nullptr;
+        for (Node * child = children->item(0); child; child = next_child)
+        {
+            next_child = child->nextSibling();
             doIncludesRecursive(config, include_from, child, zk_node_cache, zk_changed_event, contributing_zk_paths);
+        }
     }
 }
 

src/Common/tests/gtest_config_processor.cpp

@@ -0,0 +1,66 @@
+#include <Common/Config/ConfigProcessor.h>
+#include <IO/WriteBufferFromFile.h>
+#include <IO/WriteHelpers.h>
+#include <Poco/Timestamp.h>
+#include <Poco/Util/XMLConfiguration.h>
+#include <base/scope_guard.h>
+#include <gtest/gtest.h>
+#include <filesystem>
+
+
+TEST(Common, ConfigProcessorManyElements)
+{
+    namespace fs = std::filesystem;
+
+    auto path = fs::path("/tmp/test_config_processor/");
+
+    fs::create_directories(path);
+    fs::create_directories(path / "config.d");
+    SCOPE_EXIT({ fs::remove_all(path); });
+
+    auto config_file = std::make_unique<Poco::File>(path / "config.xml");
+
+    /// TODO: increase this to 1M when https://github.com/pocoproject/poco/pull/3885 is fixed
+    constexpr size_t element_count = 50000;
+
+    {
+        DB::WriteBufferFromFile out(config_file->path());
+        writeString("<clickhouse>\n", out);
+        for (size_t i = 0; i < element_count; ++i)
+            writeString("<x><name>" + std::to_string(i) + "</name></x>\n", out);
+        writeString("</clickhouse>\n", out);
+    }
+
+    Poco::Timestamp load_start;
+
+    DB::ConfigProcessor processor(config_file->path(), /* throw_on_bad_incl = */ false, /* log_to_console = */ false);
+    bool has_zk_includes;
+    DB::XMLDocumentPtr config_xml = processor.processConfig(&has_zk_includes);
+    DB::ConfigurationPtr configuration(new Poco::Util::XMLConfiguration(config_xml));
+
+    float load_elapsed_ms = (Poco::Timestamp() - load_start) / 1000.0f;
+    std::cerr << "Config loading took " << load_elapsed_ms << " ms" << std::endl;
+
+    ASSERT_EQ("0", configuration->getString("x.name"));
+    ASSERT_EQ("1", configuration->getString("x[1].name"));
+    constexpr size_t last = element_count - 1;
+    ASSERT_EQ(std::to_string(last), configuration->getString("x[" + std::to_string(last) + "].name"));
+
+    /// More that 5 min is way too slow
+    ASSERT_LE(load_elapsed_ms, 300*1000);
+
+    Poco::Timestamp enumerate_start;
+
+    Poco::Util::AbstractConfiguration::Keys keys;
+    configuration->keys("", keys);
+
+    float enumerate_elapsed_ms = (Poco::Timestamp() - enumerate_start) / 1000.0f;
+    std::cerr << "Key enumeration took " << enumerate_elapsed_ms << " ms" << std::endl;
+
+    ASSERT_EQ(element_count, keys.size());
+    ASSERT_EQ("x", keys[0]);
+    ASSERT_EQ("x[1]", keys[1]);
+
+    /// More that 5 min is way too slow
+    ASSERT_LE(enumerate_elapsed_ms, 300*1000);
+}