Merge branch 'develop' into feature/issue-231-binary-release

cmake/autogenerated_versions.txt

@@ -5,8 +5,8 @@
 SET(VERSION_REVISION 1)
 SET(VERSION_MAJOR 1)
 SET(VERSION_MINOR 3)
-SET(VERSION_PATCH 19)
-SET(VERSION_GITHASH 4436e3295c009924ec76fdeb53e2a5977122d2ef)
-SET(VERSION_DESCRIBE v1.3.19)
-SET(VERSION_STRING 1.3.19)
+SET(VERSION_PATCH 20)
+SET(VERSION_GITHASH 165325775d316ea7597f1651a0d66dd41d2183a0)
+SET(VERSION_DESCRIBE v1.3.20)
+SET(VERSION_STRING 1.3.20)
 # end of autochange

debian/changelog

@@ -1,5 +1,5 @@
-proton (1.3.19) unstable; urgency=low
+proton (1.3.20) unstable; urgency=low
 
   * Modified source code
 
- -- proton-robot <[email protected]>  Fri, 10 Nov 2023 22:59:44 +0000
+ -- proton-robot <[email protected]>  Sun, 12 Nov 2023 02:41:58 +0000

docker/client/Dockerfile

@@ -5,7 +5,7 @@ ARG apt_archive="http://archive.ubuntu.com"
 RUN sed -i "s|http://archive.ubuntu.com|$apt_archive|g" /etc/apt/sources.list
 
 ARG repository="deb https://repo.clickhouse.com/deb/stable/ main/"
-ARG version=1.3.19
+ARG version=1.3.20
 
 
 RUN apt-get update \

docker/server/Dockerfile

@@ -3,7 +3,7 @@ FROM ubuntu:20.04
 # FIXME, XXX, timeplus deb repo
 # ARG repository="deb https://repo.timeplus/deb/stable/ main/"
 ARG repository=""
-ARG version=1.3.19
+ARG version=1.3.20
 ARG gosu_ver=1.10
 
 # set non-empty deb_location_url url to create a docker image

docker/test/Dockerfile

@@ -1,7 +1,7 @@
 FROM ubuntu:18.04
 
 ARG repository="deb https://repo.clickhouse.com/deb/stable/ main/"
-ARG version=1.3.19
+ARG version=1.3.20
 
 RUN apt-get update && \
     apt-get install -y apt-transport-https dirmngr && \

programs/server/embedded.xml

@@ -22,19 +22,62 @@
     <mlock_executable>true</mlock_executable>
 
     <users>
-        <default>
-            <password></password>
+      <default>
+        <password></password>
 
-            <networks>
-                <ip>::/0</ip>
-            </networks>
+        <networks>
+          <ip>::/0</ip>
+        </networks>
 
-            <profile>default</profile>
-            <quota>default</quota>
-            <access_management>1</access_management>
-        </default>
+        <profile>default</profile>
+        <quota>default</quota>
+        <allow_databases>
+          <database>default</database>
+        </allow_databases>
+        <default_database>default</default_database>
+      </default>
+      <system>
+        <password>sys@t+</password>
+        <allow_databases>
+          <database>system</database>
+        </allow_databases>
+        <default_database>system</default_database>
+      </system>
+      <pgadmin>
+        <allow_databases>
+          <database>default</database>
+        </allow_databases>
+        <default_database>default</default_database>
+        <password>pgadmin</password>
+        <networks>
+          <ip>::/0</ip>
+        </networks>
+      </pgadmin>
+      <neutron>
+        <allow_databases>
+          <database>neutron</database>
+        </allow_databases>
+        <default_database>neutron</default_database>
+        <password>neutron@t+</password>
+        <networks>
+          <ip>::/0</ip>
+        </networks>
+      </neutron>
+      <proton>
+        <default_database>default</default_database>
+        <password>proton@t+</password>
+        <networks>
+          <ip>::1</ip>
+        </networks>
+        <networks>
+          <ip>127.0.0.1</ip>
+        </networks>
+        <access_management>1</access_management>
+      </proton>
     </users>
 
+    <grok_patterns_file>grok-patterns</grok_patterns_file>
+
     <profiles>
         <default/>
     </profiles>

src/Common/Config/ConfigReloader.cpp

@@ -143,10 +143,10 @@ void ConfigReloader::reloadIfNewer(bool force, bool throw_on_error, bool /*fallb
 struct ConfigReloader::FileWithTimestamp
 {
     std::string path;
-    time_t modification_time;
+    fs::file_time_type modification_time;
 
-    FileWithTimestamp(const std::string & path_, time_t modification_time_)
-        : path(path_), modification_time(modification_time_) {}
+    explicit FileWithTimestamp(const std::string & path_)
+        : path(path_), modification_time(fs::last_write_time(path_)) {}
 
     bool operator < (const FileWithTimestamp & rhs) const
     {
@@ -163,7 +163,7 @@ struct ConfigReloader::FileWithTimestamp
 void ConfigReloader::FilesChangesTracker::addIfExists(const std::string & path_to_add)
 {
     if (!path_to_add.empty() && fs::exists(path_to_add))
-        files.emplace(path_to_add, FS::getModificationTime(path_to_add));
+        files.emplace(path_to_add);
 }
 
 bool ConfigReloader::FilesChangesTracker::isDifferOrNewerThan(const FilesChangesTracker & rhs)

src/Common/Config/ExternalGrokPatterns.cpp

@@ -86,16 +86,34 @@ void ExternalGrokPatterns::loadPatternsFromFile()
     std::ifstream ifs(file_name);
     if (!ifs)
     {
-        LOG_WARNING(log, "External grok patterns file '{}' is not exist", file_name);
-        patterns = std::make_unique<std::unordered_map<String, String>>();
+        LOG_WARNING(log, "External grok patterns file '{}' does not exist, trying embedded resource", file_name);
+
+        /// Try to load the patterns from embedded resource
+        auto resource_data = getResource("grok-patterns");
+        if (!resource_data.empty())
+        {
+            std::string resource_string(resource_data);
+            std::istringstream resource_stream(resource_string);
+            loadPatternsFromStream(resource_stream);
+        }
+        else
+        {
+            LOG_ERROR(log, "Failed to load grok patterns from both file and embedded resource");
+            patterns = std::make_unique<std::unordered_map<String, String>>();
+        }
         return;
     }
 
+    loadPatternsFromStream(ifs);
+}
+
+void ExternalGrokPatterns::loadPatternsFromStream(std::istream & stream)
+{
     int line_num = 0;
     String line;
 
     auto new_patterns = std::make_unique<std::unordered_map<String, String>>();
-    while (std::getline(ifs, line))
+    while (std::getline(stream, line))
     {
         Poco::trimInPlace(line);
         ++line_num;

src/Common/Config/ExternalGrokPatterns.h

@@ -1,6 +1,7 @@
 #pragma once
 
 #include <Core/BackgroundSchedulePool.h>
+#include <Common/getResource.h>
 #include <Common/logger_useful.h>
 
 namespace DB
@@ -37,6 +38,7 @@ class ExternalGrokPatterns
     ExternalGrokPatterns(ContextPtr context_);
     void reloadPatternsFromFile();
     void loadPatternsFromFile();
+    void loadPatternsFromStream(std::istream & stream);
 
 private:
     std::atomic_flag is_shutdown;

src/Core/Settings.h

@@ -781,6 +781,7 @@ static constexpr UInt64 operator""_GiB(unsigned long long value)
     M(UInt64, checkpoint_interval, 0, "Checkpoint interval in seconds", 0) \
     M(UInt64, javascript_uda_max_concurrency, 1, "Control the concurrency of JavaScript UDA in a query", 0) \
     M(Float, replay_speed, 0., "Control the replay speed..0 < replay_speed < 1, means replay slower.replay_speed == 1, means replay by actual ingest interval.1 < replay_speed < <max_limit>, means replay faster", 0) \
+    M(UInt64, max_events, 0, "Total events to generate for random stream", 0) \
     /** proton: ends. */
 // End of FORMAT_FACTORY_SETTINGS
 // Please add settings non-related to formats into the COMMON_SETTINGS above.

src/Core/SortDescription.h

@@ -46,13 +46,13 @@ struct SortColumnDescription
     FillColumnDescription fill_description;
 
     explicit SortColumnDescription(
-        const std::string & column_name_,
+        std::string column_name_,
         int direction_ = 1,
         int nulls_direction_ = 1,
         const std::shared_ptr<Collator> & collator_ = nullptr,
         bool with_fill_ = false,
         const FillColumnDescription & fill_description_ = {})
-        : column_name(column_name_)
+        : column_name(std::move(column_name_))
         , direction(direction_)
         , nulls_direction(nulls_direction_)
         , collator(collator_)

src/Interpreters/Aggregator.cpp

@@ -304,7 +304,7 @@ std::optional<HashTablesCacheStatistics> getHashTablesCacheStatistics()
 void AggregatedDataVariants::convertToTwoLevel()
 {
     if (aggregator)
-        LOG_TRACE(aggregator->log, "Converting aggregation data to two-level.");
+        LOG_INFO(aggregator->log, "Converting aggregation data to two-level.");
 
     switch (type)
     {
@@ -1423,7 +1423,7 @@ void Aggregator::writeToTemporaryFile(AggregatedDataVariants & data_variants, co
     CompressedWriteBuffer compressed_buf(file_buf);
     NativeWriter block_out(compressed_buf, getHeader(false), ProtonRevision::getVersionRevision());
 
-    LOG_DEBUG(log, "Writing part of aggregation data into temporary file {}.", path);
+    LOG_INFO(log, "Writing part of aggregation data into temporary file {}.", path);
     ProfileEvents::increment(ProfileEvents::ExternalAggregationWritePart);
 
     /// Flush only two-level data and possibly overflow data.
@@ -1467,7 +1467,7 @@ void Aggregator::writeToTemporaryFile(AggregatedDataVariants & data_variants, co
     ProfileEvents::increment(ProfileEvents::ExternalAggregationCompressedBytes, compressed_bytes);
     ProfileEvents::increment(ProfileEvents::ExternalAggregationUncompressedBytes, uncompressed_bytes);
 
-    LOG_DEBUG(log,
+    LOG_INFO(log,
         "Written part in {:.3f} sec., {} rows, {} uncompressed, {} compressed,"
         " {:.3f} uncompressed bytes per row, {:.3f} compressed bytes per row, compression rate: {:.3f}"
         " ({:.3f} rows/sec., {}/sec. uncompressed, {}/sec. compressed)",
@@ -1514,6 +1514,29 @@ Block Aggregator::convertOneBucketToBlock(
     return block;
 }
 
+/// proton : starts
+/// Convert one bucket of a two level hash table data variants to a block.
+/// Use by shuffled aggregation
+Block Aggregator::convertOneBucketToBlock(AggregatedDataVariants & data_variants, Arena * arena, bool final, Int32 bucket) const
+{
+    assert(data_variants.isTwoLevel());
+
+    auto method = data_variants.type;
+    Block block;
+
+    if (false) {} // NOLINT
+#define M(NAME) \
+    else if (method == AggregatedDataVariants::Type::NAME) \
+    { \
+        block = convertOneBucketToBlock(data_variants, *data_variants.NAME, arena, final, bucket); \
+    }
+
+    APPLY_FOR_VARIANTS_TWO_LEVEL(M)
+#undef M
+
+    return block;
+}
+
 Block Aggregator::mergeAndConvertOneBucketToBlock(
     ManyAggregatedDataVariants & variants,
     Arena * arena,

src/Interpreters/Aggregator.h

@@ -1095,6 +1095,7 @@ class Aggregator final
     friend struct AggregatedDataVariants;
     friend class ConvertingAggregatedToChunksTransform;
     friend class ConvertingAggregatedToChunksSource;
+    friend class ConvertingAggregatedToChunksSourceShuffled;
     friend class AggregatingInOrderTransform;
 
     Params params;
@@ -1312,6 +1313,12 @@ class Aggregator final
         bool final,
         Int32 bucket) const;
 
+    Block convertOneBucketToBlock(
+        AggregatedDataVariants & data_variants,
+        Arena * arena,
+        bool final,
+        Int32 bucket) const;
+
     Block mergeAndConvertOneBucketToBlock(
         ManyAggregatedDataVariants & variants,
         Arena * arena,