Merge remote-tracking branch 'upstream/master' into nikvas0/bloom_filter_index

dbms/src/Common/Allocator.cpp

@@ -5,6 +5,7 @@
 #endif
 
 #include <cstdlib>
+#include <algorithm>
 #include <sys/mman.h>
 
 #include <common/mremap.h>
@@ -118,9 +119,11 @@ void * Allocator<clear_memory_>::realloc(void * buf, size_t old_size, size_t new
     if (old_size == new_size)
     {
         /// nothing to do.
+        /// BTW, it's not possible to change alignment while doing realloc.
     }
     else if (old_size < MMAP_THRESHOLD && new_size < MMAP_THRESHOLD && alignment <= MALLOC_MIN_ALIGNMENT)
     {
+        /// Resize malloc'd memory region with no special alignment requirement.
         CurrentMemoryTracker::realloc(old_size, new_size);
 
         void * new_buf = ::realloc(buf, new_size);
@@ -133,6 +136,7 @@ void * Allocator<clear_memory_>::realloc(void * buf, size_t old_size, size_t new
     }
     else if (old_size >= MMAP_THRESHOLD && new_size >= MMAP_THRESHOLD)
     {
+        /// Resize mmap'd memory region.
         CurrentMemoryTracker::realloc(old_size, new_size);
 
         // On apple and freebsd self-implemented mremap used (common/mremap.h)
@@ -142,21 +146,12 @@ void * Allocator<clear_memory_>::realloc(void * buf, size_t old_size, size_t new
 
         /// No need for zero-fill, because mmap guarantees it.
     }
-    else if (old_size >= MMAP_THRESHOLD && new_size < MMAP_THRESHOLD)
-    {
-        void * new_buf = alloc(new_size, alignment);
-        memcpy(new_buf, buf, new_size);
-        if (0 != munmap(buf, old_size))
-        {
-            ::free(new_buf);
-            DB::throwFromErrno("Allocator: Cannot munmap " + formatReadableSizeWithBinarySuffix(old_size) + ".", DB::ErrorCodes::CANNOT_MUNMAP);
-        }
-        buf = new_buf;
-    }
     else
     {
+        /// All other cases that requires a copy. MemoryTracker is called inside 'alloc', 'free' methods.
+
         void * new_buf = alloc(new_size, alignment);
-        memcpy(new_buf, buf, old_size);
+        memcpy(new_buf, buf, std::min(old_size, new_size));
         free(buf, old_size);
         buf = new_buf;
     }

dbms/src/Common/CompactArray.h

@@ -55,6 +55,28 @@ public:
         return locus;
     }
 
+    /// Used only in arcadia/metrika
+    void readText(ReadBuffer & in)
+    {
+        for (size_t i = 0; i < BITSET_SIZE; ++i)
+        {
+            if (i != 0)
+                assertChar(',', in);
+            readIntText(bitset[i], in);
+        }
+    }
+
+    /// Used only in arcadia/metrika
+    void writeText(WriteBuffer & out) const
+    {
+        for (size_t i = 0; i < BITSET_SIZE; ++i)
+        {
+            if (i != 0)
+                writeCString(",", out);
+            writeIntText(bitset[i], out);
+        }
+    }
+
 private:
     /// number of bytes in bitset
     static constexpr size_t BITSET_SIZE = (static_cast<size_t>(bucket_count) * content_width + 7) / 8;

dbms/src/Interpreters/ExternalDictionaries.cpp

@@ -36,7 +36,7 @@ ExternalDictionaries::ExternalDictionaries(
                          std::move(config_repository),
                          &Logger::get("ExternalDictionaries"),
                          "external dictionary"),
-          context(context)
+        context(context)
 {
     init(throw_on_error);
 }

dbms/src/Interpreters/ExternalLoader.cpp

@@ -155,14 +155,6 @@ void ExternalLoader::reloadAndUpdate(bool throw_on_error)
     /// periodic update
     std::vector<std::pair<std::string, LoadablePtr>> objects_to_update;
 
-    auto getNextUpdateTime = [this](const LoadablePtr & current)
-    {
-        /// calculate next update time
-        const auto & lifetime = current->getLifetime();
-        std::uniform_int_distribution<UInt64> distribution{lifetime.min_sec, lifetime.max_sec};
-        return std::chrono::system_clock::now() + std::chrono::seconds{distribution(rnd_engine)};
-    };
-
     /// Collect objects that needs to be updated under lock. Then create new versions without lock, and assign under lock.
     {
         std::lock_guard lock{map_mutex};
@@ -185,18 +177,12 @@ void ExternalLoader::reloadAndUpdate(bool throw_on_error)
                 if (!current->supportUpdates())
                     continue;
 
-                auto & update_time = update_times[current->getName()];
+                auto update_time = update_times[current->getName()];
 
                 /// check that timeout has passed
                 if (std::chrono::system_clock::now() < update_time)
                     continue;
 
-                if (!current->isModified())
-                {
-                    update_time = getNextUpdateTime(current);
-                    continue;
-                }
-
                 objects_to_update.emplace_back(loadable_object.first, current);
             }
             catch (...)
@@ -209,6 +195,14 @@ void ExternalLoader::reloadAndUpdate(bool throw_on_error)
         }
     }
 
+    auto getNextUpdateTime = [this](const LoadablePtr & current)
+    {
+        /// Calculate next update time.
+        const auto & lifetime = current->getLifetime();
+        std::uniform_int_distribution<UInt64> distribution{lifetime.min_sec, lifetime.max_sec};
+        return std::chrono::system_clock::now() + std::chrono::seconds{distribution(rnd_engine)};
+    };
+
     for (auto & [name, current] : objects_to_update)
     {
         LoadablePtr new_version;
@@ -216,9 +210,12 @@ void ExternalLoader::reloadAndUpdate(bool throw_on_error)
 
         try
         {
-            /// create new version of loadable object
-            new_version = current->clone();
-            exception = new_version->getCreationException();
+            if (current->isModified())
+            {
+                /// Create new version of loadable object.
+                new_version = current->clone();
+                exception = new_version->getCreationException();
+            }
         }
         catch (...)
         {
@@ -235,8 +232,12 @@ void ExternalLoader::reloadAndUpdate(bool throw_on_error)
                 it->second.exception = exception;
                 if (!exception)
                 {
-                    it->second.loadable.reset();
-                    it->second.loadable = std::move(new_version);
+                    /// If the dictionary is not modified - leave old version.
+                    if (new_version)
+                    {
+                        it->second.loadable.reset();
+                        it->second.loadable = std::move(new_version);
+                    }
                 }
                 else
                 {

dbms/src/Interpreters/Set.h

@@ -192,4 +192,4 @@ private:
     std::vector<KeyTuplePositionMapping> indexes_mapping;
 };
 
- }
+}

dbms/src/Storages/MergeTree/MergeTreeSequentialBlockInputStream.cpp

@@ -26,7 +26,7 @@ MergeTreeSequentialBlockInputStream::MergeTreeSequentialBlockInputStream(
     {
         std::stringstream message;
         message << "Reading " << data_part->marks_count << " marks from part " << data_part->name
-            << ", totaly " << data_part->rows_count
+            << ", total " << data_part->rows_count
             << " rows starting from the beginning of the part, columns: ";
         for (size_t i = 0, size = columns_to_read.size(); i < size; ++i)
             message << (i == 0 ? "" : ", ") << columns_to_read[i];

libs/libdaemon/src/BaseDaemon.cpp

@@ -700,12 +700,12 @@ static void checkRequiredInstructions(volatile InstructionFail & fail)
 
 #if __AVX2__
     fail = InstructionFail::AVX2;
-    __asm__ volatile ("vpabsw %%ymm0, %%ymm0, %%ymm0" : : : "ymm0");
+    __asm__ volatile ("vpabsw %%ymm0, %%ymm0" : : : "ymm0");
 #endif
 
 #if __AVX512__
     fail = InstructionFail::AVX512;
-    __asm__ volatile ("vpabsw %%zmm0, %%zmm0, %%zmm0" : : : "zmm0");
+    __asm__ volatile ("vpabsw %%zmm0, %%zmm0" : : : "zmm0");
 #endif
 
     fail = InstructionFail::NONE;