Merge remote-tracking branch 'origin/master' into pr-right-joins

CMakeLists.txt

@@ -88,6 +88,7 @@ string (TOUPPER ${CMAKE_BUILD_TYPE} CMAKE_BUILD_TYPE_UC)
 list(REVERSE CMAKE_FIND_LIBRARY_SUFFIXES)
 
 option (ENABLE_FUZZING "Fuzzy testing using libfuzzer" OFF)
+option (ENABLE_FUZZER_TEST "Build testing fuzzers in order to test libFuzzer functionality" OFF)
 
 if (ENABLE_FUZZING)
     # Also set WITH_COVERAGE=1 for better fuzzing process

SECURITY.md

@@ -14,9 +14,10 @@ The following versions of ClickHouse server are currently supported with securit
 
 | Version | Supported |
 |:-|:-|
+| 24.10 | ✔️ |
 | 24.9 | ✔️ |
 | 24.8 | ✔️ |
-| 24.7 | ✔️ |
+| 24.7 | ❌ |
 | 24.6 | ❌ |
 | 24.5 | ❌ |
 | 24.4 | ❌ |

docker/keeper/Dockerfile

@@ -34,7 +34,7 @@ RUN arch=${TARGETARCH:-amd64} \
 # lts / testing / prestable / etc
 ARG REPO_CHANNEL="stable"
 ARG REPOSITORY="https://packages.clickhouse.com/tgz/${REPO_CHANNEL}"
-ARG VERSION="24.9.2.42"
+ARG VERSION="24.10.1.2812"
 ARG PACKAGES="clickhouse-keeper"
 ARG DIRECT_DOWNLOAD_URLS=""
 

docker/server/Dockerfile.alpine

@@ -35,7 +35,7 @@ RUN arch=${TARGETARCH:-amd64} \
 # lts / testing / prestable / etc
 ARG REPO_CHANNEL="stable"
 ARG REPOSITORY="https://packages.clickhouse.com/tgz/${REPO_CHANNEL}"
-ARG VERSION="24.9.2.42"
+ARG VERSION="24.10.1.2812"
 ARG PACKAGES="clickhouse-client clickhouse-server clickhouse-common-static"
 ARG DIRECT_DOWNLOAD_URLS=""
 

docker/server/Dockerfile.ubuntu

@@ -28,7 +28,7 @@ RUN sed -i "s|http://archive.ubuntu.com|${apt_archive}|g" /etc/apt/sources.list
 
 ARG REPO_CHANNEL="stable"
 ARG REPOSITORY="deb [signed-by=/usr/share/keyrings/clickhouse-keyring.gpg] https://packages.clickhouse.com/deb ${REPO_CHANNEL} main"
-ARG VERSION="24.9.2.42"
+ARG VERSION="24.10.1.2812"
 ARG PACKAGES="clickhouse-client clickhouse-server clickhouse-common-static"
 
 #docker-official-library:off

docker/test/libfuzzer/Dockerfile

@@ -33,8 +33,6 @@ RUN apt-get update \
 COPY requirements.txt /
 RUN pip3 install --no-cache-dir -r /requirements.txt
 
-ENV FUZZER_ARGS="-max_total_time=60"
-
 SHELL ["/bin/bash", "-c"]
 
 # docker run --network=host --volume <workspace>:/workspace -e PR_TO_TEST=<> -e SHA_TO_TEST=<> clickhouse/libfuzzer

programs/server/Server.cpp

@@ -1353,9 +1353,11 @@ try
     }
 
     FailPointInjection::enableFromGlobalConfig(config());
+#endif
 
     memory_worker.start();
 
+#if defined(OS_LINUX)
     int default_oom_score = 0;
 
 #if !defined(NDEBUG)

src/Access/AccessControl.cpp

@@ -608,7 +608,7 @@ AuthResult AccessControl::authenticate(const Credentials & credentials, const Po
     }
     catch (...)
     {
-        tryLogCurrentException(getLogger(), "from: " + address.toString() + ", user: " + credentials.getUserName()  + ": Authentication failed");
+        tryLogCurrentException(getLogger(), "from: " + address.toString() + ", user: " + credentials.getUserName()  + ": Authentication failed", LogsLevel::information);
 
         WriteBufferFromOwnString message;
         message << credentials.getUserName() << ": Authentication failed: password is incorrect, or there is no user with such name.";
@@ -622,8 +622,9 @@ AuthResult AccessControl::authenticate(const Credentials & credentials, const Po
                 << "and deleting this file will reset the password.\n"
                 << "See also /etc/clickhouse-server/users.xml on the server where ClickHouse is installed.\n\n";
 
-        /// We use the same message for all authentication failures because we don't want to give away any unnecessary information for security reasons,
-        /// only the log will show the exact reason.
+        /// We use the same message for all authentication failures because we don't want to give away any unnecessary information for security reasons.
+        /// Only the log ((*), above) will show the exact reason. Note that (*) logs at information level instead of the default error level as
+        /// authentication failures are not an unusual event.
         throw Exception(PreformattedMessage{message.str(),
                                             "{}: Authentication failed: password is incorrect, or there is no user with such name",
                                             std::vector<std::string>{credentials.getUserName()}},

src/AggregateFunctions/fuzzers/CMakeLists.txt

@@ -1,2 +1,2 @@
 clickhouse_add_executable(aggregate_function_state_deserialization_fuzzer aggregate_function_state_deserialization_fuzzer.cpp ${SRCS})
-target_link_libraries(aggregate_function_state_deserialization_fuzzer PRIVATE clickhouse_aggregate_functions)
+target_link_libraries(aggregate_function_state_deserialization_fuzzer PRIVATE clickhouse_aggregate_functions dbms)

src/Common/Exception.cpp

@@ -251,7 +251,7 @@ void Exception::setThreadFramePointers(ThreadFramePointersBase frame_pointers)
         thread_frame_pointers.frame_pointers = std::move(frame_pointers);
 }
 
-static void tryLogCurrentExceptionImpl(Poco::Logger * logger, const std::string & start_of_message)
+static void tryLogCurrentExceptionImpl(Poco::Logger * logger, const std::string & start_of_message, LogsLevel level)
 {
     if (!isLoggingEnabled())
         return;
@@ -262,14 +262,25 @@ static void tryLogCurrentExceptionImpl(Poco::Logger * logger, const std::string
         if (!start_of_message.empty())
             message.text = fmt::format("{}: {}", start_of_message, message.text);
 
-        LOG_ERROR(logger, message);
+        switch (level)
+        {
+            case LogsLevel::none: break;
+            case LogsLevel::test: LOG_TEST(logger, message); break;
+            case LogsLevel::trace: LOG_TRACE(logger, message); break;
+            case LogsLevel::debug: LOG_DEBUG(logger, message); break;
+            case LogsLevel::information: LOG_INFO(logger, message); break;
+            case LogsLevel::warning: LOG_WARNING(logger, message); break;
+            case LogsLevel::error: LOG_ERROR(logger, message); break;
+            case LogsLevel::fatal: LOG_FATAL(logger, message); break;
+        }
+
     }
     catch (...) // NOLINT(bugprone-empty-catch)
     {
     }
 }
 
-void tryLogCurrentException(const char * log_name, const std::string & start_of_message)
+void tryLogCurrentException(const char * log_name, const std::string & start_of_message, LogsLevel level)
 {
     if (!isLoggingEnabled())
         return;
@@ -283,10 +294,10 @@ void tryLogCurrentException(const char * log_name, const std::string & start_of_
 
     /// getLogger can allocate memory too
     auto logger = getLogger(log_name);
-    tryLogCurrentExceptionImpl(logger.get(), start_of_message);
+    tryLogCurrentExceptionImpl(logger.get(), start_of_message, level);
 }
 
-void tryLogCurrentException(Poco::Logger * logger, const std::string & start_of_message)
+void tryLogCurrentException(Poco::Logger * logger, const std::string & start_of_message, LogsLevel level)
 {
     /// Under high memory pressure, new allocations throw a
     /// MEMORY_LIMIT_EXCEEDED exception.
@@ -295,17 +306,17 @@ void tryLogCurrentException(Poco::Logger * logger, const std::string & start_of_
     /// MemoryTracker until the exception will be logged.
     LockMemoryExceptionInThread lock_memory_tracker(VariableContext::Global);
 
-    tryLogCurrentExceptionImpl(logger, start_of_message);
+    tryLogCurrentExceptionImpl(logger, start_of_message, level);
 }
 
-void tryLogCurrentException(LoggerPtr logger, const std::string & start_of_message)
+void tryLogCurrentException(LoggerPtr logger, const std::string & start_of_message, LogsLevel level)
 {
-    tryLogCurrentException(logger.get(), start_of_message);
+    tryLogCurrentException(logger.get(), start_of_message, level);
 }
 
-void tryLogCurrentException(const AtomicLogger & logger, const std::string & start_of_message)
+void tryLogCurrentException(const AtomicLogger & logger, const std::string & start_of_message, LogsLevel level)
 {
-    tryLogCurrentException(logger.load(), start_of_message);
+    tryLogCurrentException(logger.load(), start_of_message, level);
 }
 
 static void getNoSpaceLeftInfoMessage(std::filesystem::path path, String & msg)

src/Common/Exception.h

@@ -7,6 +7,7 @@
 #include <Common/Logger.h>
 #include <Common/LoggingFormatStringHelpers.h>
 #include <Common/StackTrace.h>
+#include <Core/LogsLevel.h>
 
 #include <cerrno>
 #include <exception>
@@ -276,10 +277,10 @@ using Exceptions = std::vector<std::exception_ptr>;
   * Can be used in destructors in the catch-all block.
   */
 /// TODO: Logger leak constexpr overload
-void tryLogCurrentException(const char * log_name, const std::string & start_of_message = "");
-void tryLogCurrentException(Poco::Logger * logger, const std::string & start_of_message = "");
-void tryLogCurrentException(LoggerPtr logger, const std::string & start_of_message = "");
-void tryLogCurrentException(const AtomicLogger & logger, const std::string & start_of_message = "");
+void tryLogCurrentException(const char * log_name, const std::string & start_of_message = "", LogsLevel level = LogsLevel::error);
+void tryLogCurrentException(Poco::Logger * logger, const std::string & start_of_message = "", LogsLevel level = LogsLevel::error);
+void tryLogCurrentException(LoggerPtr logger, const std::string & start_of_message = "", LogsLevel level = LogsLevel::error);
+void tryLogCurrentException(const AtomicLogger & logger, const std::string & start_of_message = "", LogsLevel level = LogsLevel::error);
 
 
 /** Prints current exception in canonical format.

src/Common/NamedCollections/NamedCollectionsMetadataStorage.cpp

@@ -568,7 +568,7 @@ std::vector<std::string> NamedCollectionsMetadataStorage::listCollections() cons
     std::vector<std::string> collections;
     collections.reserve(paths.size());
     for (const auto & path : paths)
-        collections.push_back(std::filesystem::path(path).stem());
+        collections.push_back(unescapeForFileName(std::filesystem::path(path).stem()));
     return collections;
 }
 

src/Common/ProfileEvents.cpp

@@ -547,6 +547,7 @@ The server successfully detected this situation and will download merged part fr
     M(FilesystemCacheLoadMetadataMicroseconds, "Time spent loading filesystem cache metadata", ValueType::Microseconds) \
     M(FilesystemCacheEvictedBytes, "Number of bytes evicted from filesystem cache", ValueType::Bytes) \
     M(FilesystemCacheEvictedFileSegments, "Number of file segments evicted from filesystem cache", ValueType::Number) \
+    M(FilesystemCacheBackgroundDownloadQueuePush, "Number of file segments sent for background download in filesystem cache", ValueType::Number) \
     M(FilesystemCacheEvictionSkippedFileSegments, "Number of file segments skipped for eviction because of being in unreleasable state", ValueType::Number) \
     M(FilesystemCacheEvictionSkippedEvictingFileSegments, "Number of file segments skipped for eviction because of being in evicting state", ValueType::Number) \
     M(FilesystemCacheEvictionTries, "Number of filesystem cache eviction attempts", ValueType::Number) \

src/Core/Settings.cpp

@@ -5132,6 +5132,12 @@ Only in ClickHouse Cloud. A window for sending ACK for DataPacket sequence in a
 )", 0) \
     DECLARE(Bool, distributed_cache_discard_connection_if_unread_data, true, R"(
 Only in ClickHouse Cloud. Discard connection if some data is unread.
+)", 0) \
+    DECLARE(Bool, filesystem_cache_enable_background_download_for_metadata_files_in_packed_storage, true, R"(
+Only in ClickHouse Cloud. Wait time to lock cache for space reservation in filesystem cache
+)", 0) \
+    DECLARE(Bool, filesystem_cache_enable_background_download_during_fetch, true, R"(
+Only in ClickHouse Cloud. Wait time to lock cache for space reservation in filesystem cache
 )", 0) \
     \
     DECLARE(Bool, parallelize_output_from_storages, true, R"(
@@ -5142,6 +5148,7 @@ The setting allows a user to provide own deduplication semantic in MergeTree/Rep
 For example, by providing a unique value for the setting in each INSERT statement,
 user can avoid the same inserted data being deduplicated.
 
+
 Possible values:
 
 - Any string

src/Core/SettingsChangesHistory.cpp

@@ -65,12 +65,15 @@ static std::initializer_list<std::pair<ClickHouseVersion, SettingsChangesHistory
     {"24.11",
         {
             {"distributed_cache_discard_connection_if_unread_data", true, true, "New setting"},
+            {"filesystem_cache_enable_background_download_for_metadata_files_in_packed_storage", true, true, "New setting"},
+            {"filesystem_cache_enable_background_download_during_fetch", true, true, "New setting"},
             {"azure_check_objects_after_upload", false, false, "Check each uploaded object in azure blob storage to be sure that upload was successful"},
             {"backup_restore_keeper_max_retries", 20, 1000, "Should be big enough so the whole operation BACKUP or RESTORE operation won't fail because of a temporary [Zoo]Keeper failure in the middle of it."},
             {"backup_restore_failure_after_host_disconnected_for_seconds", 0, 3600, "New setting."},
             {"backup_restore_keeper_max_retries_while_initializing", 0, 20, "New setting."},
             {"backup_restore_keeper_max_retries_while_handling_error", 0, 20, "New setting."},
             {"backup_restore_finish_timeout_after_error_sec", 0, 180, "New setting."},
+            {"query_plan_join_inner_table_selection", "auto", "auto", "New setting."},
         }
     },
     {"24.10",
@@ -85,7 +88,6 @@ static std::initializer_list<std::pair<ClickHouseVersion, SettingsChangesHistory
             {"restore_replace_external_dictionary_source_to_null", false, false, "New setting."},
             {"show_create_query_identifier_quoting_rule", "when_necessary", "when_necessary", "New setting."},
             {"show_create_query_identifier_quoting_style", "Backticks", "Backticks", "New setting."},
-            {"query_plan_join_inner_table_selection", "auto", "auto", "New setting."},
             {"merge_tree_min_read_task_size", 8, 8, "New setting"},
             {"merge_tree_min_rows_for_concurrent_read_for_remote_filesystem", (20 * 8192), 0, "Setting is deprecated"},
             {"merge_tree_min_bytes_for_concurrent_read_for_remote_filesystem", (24 * 10 * 1024 * 1024), 0, "Setting is deprecated"},

src/Core/fuzzers/CMakeLists.txt

@@ -1,2 +1,2 @@
 clickhouse_add_executable (names_and_types_fuzzer names_and_types_fuzzer.cpp)
-target_link_libraries (names_and_types_fuzzer PRIVATE)
+target_link_libraries (names_and_types_fuzzer PRIVATE dbms)

src/DataTypes/fuzzers/CMakeLists.txt

@@ -1,2 +1,3 @@
 clickhouse_add_executable(data_type_deserialization_fuzzer data_type_deserialization_fuzzer.cpp ${SRCS})
-target_link_libraries(data_type_deserialization_fuzzer PRIVATE clickhouse_aggregate_functions)
+
+target_link_libraries(data_type_deserialization_fuzzer PRIVATE clickhouse_aggregate_functions dbms)

src/DataTypes/fuzzers/data_type_deserialization_fuzzer.cpp

@@ -3,6 +3,7 @@
 #include <IO/ReadBufferFromMemory.h>
 #include <IO/ReadHelpers.h>
 
+#include <DataTypes/IDataType.h>
 #include <DataTypes/DataTypeFactory.h>
 
 #include <Common/MemoryTracker.h>

src/Disks/IO/CachedOnDiskReadBufferFromFile.cpp

@@ -535,7 +535,7 @@ bool CachedOnDiskReadBufferFromFile::completeFileSegmentAndGetNext()
     chassert(file_offset_of_buffer_end > completed_range.right);
     cache_file_reader.reset();
 
-    file_segments->popFront();
+    file_segments->completeAndPopFront(settings.filesystem_cache_allow_background_download);
     if (file_segments->empty() && !nextFileSegmentsBatch())
         return false;
 
@@ -556,6 +556,12 @@ CachedOnDiskReadBufferFromFile::~CachedOnDiskReadBufferFromFile()
     {
         appendFilesystemCacheLog(file_segments->front(), read_type);
     }
+
+    if (file_segments && !file_segments->empty() && !file_segments->front().isCompleted())
+    {
+        file_segments->completeAndPopFront(settings.filesystem_cache_allow_background_download);
+        file_segments = {};
+    }
 }
 
 void CachedOnDiskReadBufferFromFile::predownload(FileSegment & file_segment)

src/Disks/IO/CachedOnDiskWriteBufferFromFile.cpp

@@ -196,7 +196,7 @@ void FileSegmentRangeWriter::completeFileSegment()
     if (file_segment.isDetached() || file_segment.isCompleted())
         return;
 
-    file_segment.complete();
+    file_segment.complete(false);
     appendFilesystemCacheLog(file_segment);
 }
 
@@ -210,7 +210,7 @@ void FileSegmentRangeWriter::jumpToPosition(size_t position)
         if (position < current_write_offset)
             throw Exception(ErrorCodes::LOGICAL_ERROR, "Cannot jump backwards: {} < {}", position, current_write_offset);
 
-        file_segment.complete();
+        file_segment.complete(false);
         file_segments.reset();
     }
     expected_write_offset = position;

src/Formats/fuzzers/CMakeLists.txt

@@ -1,2 +1,2 @@
 clickhouse_add_executable(format_fuzzer format_fuzzer.cpp ${SRCS})
-target_link_libraries(format_fuzzer PRIVATE clickhouse_aggregate_functions)
+target_link_libraries(format_fuzzer PRIVATE clickhouse_aggregate_functions dbms)

src/IO/ReadSettings.h

@@ -58,6 +58,9 @@ struct ReadSettings
     bool enable_filesystem_cache_log = false;
     size_t filesystem_cache_segments_batch_size = 20;
     size_t filesystem_cache_reserve_space_wait_lock_timeout_milliseconds = 1000;
+    bool filesystem_cache_allow_background_download = true;
+    bool filesystem_cache_allow_background_download_for_metadata_files_in_packed_storage = true;
+    bool filesystem_cache_allow_background_download_during_fetch = true;
 
     bool use_page_cache_for_disks_without_file_cache = false;
     bool read_from_page_cache_if_exists_otherwise_bypass_cache = false;

src/Interpreters/Cache/FileSegment.cpp

@@ -28,6 +28,7 @@ namespace ProfileEvents
     extern const Event FileSegmentFailToIncreasePriority;
     extern const Event FilesystemCacheHoldFileSegments;
     extern const Event FilesystemCacheUnusedHoldFileSegments;
+    extern const Event FilesystemCacheBackgroundDownloadQueuePush;
 }
 
 namespace CurrentMetrics
@@ -627,7 +628,7 @@ void FileSegment::completePartAndResetDownloader()
     LOG_TEST(log, "Complete batch. ({})", getInfoForLogUnlocked(lk));
 }
 
-void FileSegment::complete()
+void FileSegment::complete(bool allow_background_download)
 {
     ProfileEventTimeIncrement<Microseconds> watch(ProfileEvents::FileSegmentCompleteMicroseconds);
 
@@ -704,8 +705,9 @@ void FileSegment::complete()
             if (is_last_holder)
             {
                 bool added_to_download_queue = false;
-                if (background_download_enabled && remote_file_reader)
+                if (allow_background_download && background_download_enabled && remote_file_reader)
                 {
+                    ProfileEvents::increment(ProfileEvents::FilesystemCacheBackgroundDownloadQueuePush);
                     added_to_download_queue = locked_key->addToDownloadQueue(offset(), segment_lock); /// Finish download in background.
                 }
 
@@ -1001,7 +1003,14 @@ void FileSegmentsHolder::reset()
 
     ProfileEvents::increment(ProfileEvents::FilesystemCacheUnusedHoldFileSegments, file_segments.size());
     for (auto file_segment_it = file_segments.begin(); file_segment_it != file_segments.end();)
-        file_segment_it = completeAndPopFrontImpl();
+    {
+        /// One might think it would have been more correct to do `false` here,
+        /// not to allow background download for file segments that we actually did not start reading.
+        /// But actually we would only do that, if those file segments were already read partially by some other thread/query
+        /// but they were not put to the download queue, because current thread was holding them in Holder.
+        /// So as a culprit, we need to allow to happen what would have happened if we did not exist.
+        file_segment_it = completeAndPopFrontImpl(true);
+    }
     file_segments.clear();
 }
 
@@ -1010,9 +1019,9 @@ FileSegmentsHolder::~FileSegmentsHolder()
     reset();
 }
 
-FileSegments::iterator FileSegmentsHolder::completeAndPopFrontImpl()
+FileSegments::iterator FileSegmentsHolder::completeAndPopFrontImpl(bool allow_background_download)
 {
-    front().complete();
+    front().complete(allow_background_download);
     CurrentMetrics::sub(CurrentMetrics::FilesystemCacheHoldFileSegments);
     return file_segments.erase(file_segments.begin());
 }

src/Interpreters/Cache/FileSegment.h

@@ -189,7 +189,7 @@ public:
      * ========== Methods that must do cv.notify() ==================
      */
 
-    void complete();
+    void complete(bool allow_background_download);
 
     void completePartAndResetDownloader();
 
@@ -297,7 +297,7 @@ struct FileSegmentsHolder final : private boost::noncopyable
 
     String toString(bool with_state = false) const;
 
-    void popFront() { completeAndPopFrontImpl(); }
+    void completeAndPopFront(bool allow_background_download) { completeAndPopFrontImpl(allow_background_download); }
 
     FileSegment & front() { return *file_segments.front(); }
     const FileSegment & front() const { return *file_segments.front(); }
@@ -319,7 +319,7 @@ struct FileSegmentsHolder final : private boost::noncopyable
 private:
     FileSegments file_segments{};
 
-    FileSegments::iterator completeAndPopFrontImpl();
+    FileSegments::iterator completeAndPopFrontImpl(bool allow_background_download);
 };
 
 using FileSegmentsHolderPtr = std::unique_ptr<FileSegmentsHolder>;

src/Interpreters/Context.cpp

@@ -194,6 +194,8 @@ namespace Setting
     extern const SettingsUInt64 filesystem_cache_max_download_size;
     extern const SettingsUInt64 filesystem_cache_reserve_space_wait_lock_timeout_milliseconds;
     extern const SettingsUInt64 filesystem_cache_segments_batch_size;
+    extern const SettingsBool filesystem_cache_enable_background_download_for_metadata_files_in_packed_storage;
+    extern const SettingsBool filesystem_cache_enable_background_download_during_fetch;
     extern const SettingsBool http_make_head_request;
     extern const SettingsUInt64 http_max_fields;
     extern const SettingsUInt64 http_max_field_name_size;
@@ -5746,6 +5748,9 @@ ReadSettings Context::getReadSettings() const
     res.filesystem_cache_segments_batch_size = settings_ref[Setting::filesystem_cache_segments_batch_size];
     res.filesystem_cache_reserve_space_wait_lock_timeout_milliseconds
         = settings_ref[Setting::filesystem_cache_reserve_space_wait_lock_timeout_milliseconds];
+    res.filesystem_cache_allow_background_download_for_metadata_files_in_packed_storage
+        = settings_ref[Setting::filesystem_cache_enable_background_download_for_metadata_files_in_packed_storage];
+    res.filesystem_cache_allow_background_download_during_fetch = settings_ref[Setting::filesystem_cache_enable_background_download_during_fetch];
 
     res.filesystem_cache_max_download_size = settings_ref[Setting::filesystem_cache_max_download_size];
     res.skip_download_if_exceeds_query_cache = settings_ref[Setting::skip_download_if_exceeds_query_cache];

src/Interpreters/fuzzers/CMakeLists.txt

@@ -3,5 +3,6 @@ target_link_libraries(execute_query_fuzzer PRIVATE
     dbms
     clickhouse_table_functions
     clickhouse_aggregate_functions
+    clickhouse_functions
     clickhouse_dictionaries
     clickhouse_dictionaries_embedded)

src/Interpreters/tests/gtest_filecache.cpp

@@ -253,7 +253,7 @@ void download(FileSegment & file_segment)
     download(cache_base_path, file_segment);
     ASSERT_EQ(file_segment.state(), State::DOWNLOADING);
 
-    file_segment.complete();
+    file_segment.complete(false);
     ASSERT_EQ(file_segment.state(), State::DOWNLOADED);
 }
 
@@ -263,7 +263,7 @@ void assertDownloadFails(FileSegment & file_segment)
     ASSERT_EQ(file_segment.getDownloadedSize(), 0);
     std::string failure_reason;
     ASSERT_FALSE(file_segment.reserve(file_segment.range().size(), 1000, failure_reason));
-    file_segment.complete();
+    file_segment.complete(false);
 }
 
 void download(const HolderPtr & holder)
@@ -971,7 +971,7 @@ TEST_F(FileCacheTest, temporaryData)
             ASSERT_TRUE(segment->getOrSetDownloader() == DB::FileSegment::getCallerId());
             ASSERT_TRUE(segment->reserve(segment->range().size(), 1000, failure_reason));
             download(*segment);
-            segment->complete();
+            segment->complete(false);
         }
     }
 

src/Parsers/fuzzers/CMakeLists.txt

@@ -2,10 +2,10 @@ clickhouse_add_executable(lexer_fuzzer lexer_fuzzer.cpp ${SRCS})
 target_link_libraries(lexer_fuzzer PRIVATE clickhouse_parsers)
 
 clickhouse_add_executable(select_parser_fuzzer select_parser_fuzzer.cpp ${SRCS})
-target_link_libraries(select_parser_fuzzer PRIVATE clickhouse_parsers dbms)
+target_link_libraries(select_parser_fuzzer PRIVATE clickhouse_parsers clickhouse_functions dbms)
 
 clickhouse_add_executable(create_parser_fuzzer create_parser_fuzzer.cpp ${SRCS})
-target_link_libraries(create_parser_fuzzer PRIVATE clickhouse_parsers dbms)
+target_link_libraries(create_parser_fuzzer PRIVATE clickhouse_parsers clickhouse_functions dbms)
 
 add_subdirectory(codegen_fuzzer)
 

src/Parsers/fuzzers/codegen_fuzzer/CMakeLists.txt

@@ -47,4 +47,4 @@ target_compile_options (codegen_select_fuzzer PRIVATE -Wno-newline-eof)
 target_link_libraries(protoc ch_contrib::fuzzer)
 
 target_include_directories(codegen_select_fuzzer SYSTEM BEFORE PRIVATE "${CMAKE_CURRENT_BINARY_DIR}")
-target_link_libraries(codegen_select_fuzzer PRIVATE ch_contrib::protobuf_mutator ch_contrib::protoc dbms)
+target_link_libraries(codegen_select_fuzzer PRIVATE ch_contrib::protobuf_mutator ch_contrib::protoc clickhouse_functions dbms)

src/Server/TCPHandler.cpp

@@ -1614,7 +1614,8 @@ void TCPHandler::receiveHello()
                 if (e.code() != DB::ErrorCodes::AUTHENTICATION_FAILED)
                     throw;
 
-                tryLogCurrentException(log, "SSL authentication failed, falling back to password authentication");
+                tryLogCurrentException(log, "SSL authentication failed, falling back to password authentication", LogsLevel::information);
+                /// ^^ Log at debug level instead of default error level as authentication failures are not an unusual event.
             }
         }
     }

src/Storages/MergeTree/DataPartsExchange.cpp

@@ -908,7 +908,7 @@ MergeTreeData::MutableDataPartPtr Fetcher::downloadPartToDisk(
     {
         part_storage_for_loading->commitTransaction();
 
-        MergeTreeDataPartBuilder builder(data, part_name, volume, part_relative_path, part_dir);
+        MergeTreeDataPartBuilder builder(data, part_name, volume, part_relative_path, part_dir, getReadSettings());
         new_data_part = builder.withPartFormatFromDisk().build();
 
         new_data_part->version.setCreationTID(Tx::PrehistoricTID, nullptr);

src/Storages/MergeTree/IMergeTreeDataPart.cpp

@@ -833,7 +833,7 @@ MergeTreeDataPartBuilder IMergeTreeDataPart::getProjectionPartBuilder(const Stri
 {
     const char * projection_extension = is_temp_projection ? ".tmp_proj" : ".proj";
     auto projection_storage = getDataPartStorage().getProjection(projection_name + projection_extension, !is_temp_projection);
-    MergeTreeDataPartBuilder builder(storage, projection_name, projection_storage);
+    MergeTreeDataPartBuilder builder(storage, projection_name, projection_storage, getReadSettings());
     return builder.withPartInfo(MergeListElement::FAKE_RESULT_PART_FOR_PROJECTION).withParentPart(this);
 }
 

src/Storages/MergeTree/MergeTask.cpp

@@ -349,13 +349,13 @@ bool MergeTask::ExecuteAndFinalizeHorizontalPart::prepare() const
     if (global_ctx->parent_part)
     {
         auto data_part_storage = global_ctx->parent_part->getDataPartStorage().getProjection(local_tmp_part_basename,  /* use parent transaction */ false);
-        builder.emplace(*global_ctx->data, global_ctx->future_part->name, data_part_storage);
+        builder.emplace(*global_ctx->data, global_ctx->future_part->name, data_part_storage, getReadSettings());
         builder->withParentPart(global_ctx->parent_part);
     }
     else
     {
         auto local_single_disk_volume = std::make_shared<SingleDiskVolume>("volume_" + global_ctx->future_part->name, global_ctx->disk, 0);
-        builder.emplace(global_ctx->data->getDataPartBuilder(global_ctx->future_part->name, local_single_disk_volume, local_tmp_part_basename));
+        builder.emplace(global_ctx->data->getDataPartBuilder(global_ctx->future_part->name, local_single_disk_volume, local_tmp_part_basename, getReadSettings()));
         builder->withPartStorageType(global_ctx->future_part->part_format.storage_type);
     }
 

src/Storages/MergeTree/MergeTreeData.cpp

@@ -1442,7 +1442,7 @@ void MergeTreeData::loadUnexpectedDataPart(UnexpectedPartLoadState & state)
 
     try
     {
-        state.part = getDataPartBuilder(part_name, single_disk_volume, part_name)
+        state.part = getDataPartBuilder(part_name, single_disk_volume, part_name, getReadSettings())
             .withPartInfo(part_info)
             .withPartFormatFromDisk()
             .build();
@@ -1457,7 +1457,7 @@ void MergeTreeData::loadUnexpectedDataPart(UnexpectedPartLoadState & state)
             /// Build a fake part and mark it as broken in case of filesystem error.
             /// If the error impacts part directory instead of single files,
             /// an exception will be thrown during detach and silently ignored.
-            state.part = getDataPartBuilder(part_name, single_disk_volume, part_name)
+            state.part = getDataPartBuilder(part_name, single_disk_volume, part_name, getReadSettings())
                 .withPartStorageType(MergeTreeDataPartStorageType::Full)
                 .withPartType(MergeTreeDataPartType::Wide)
                 .build();
@@ -1491,7 +1491,7 @@ MergeTreeData::LoadPartResult MergeTreeData::loadDataPart(
             /// Build a fake part and mark it as broken in case of filesystem error.
             /// If the error impacts part directory instead of single files,
             /// an exception will be thrown during detach and silently ignored.
-            res.part = getDataPartBuilder(part_name, single_disk_volume, part_name)
+            res.part = getDataPartBuilder(part_name, single_disk_volume, part_name, getReadSettings())
                 .withPartStorageType(MergeTreeDataPartStorageType::Full)
                 .withPartType(MergeTreeDataPartType::Wide)
                 .build();
@@ -1512,7 +1512,7 @@ MergeTreeData::LoadPartResult MergeTreeData::loadDataPart(
 
     try
     {
-        res.part = getDataPartBuilder(part_name, single_disk_volume, part_name)
+        res.part = getDataPartBuilder(part_name, single_disk_volume, part_name, getReadSettings())
             .withPartInfo(part_info)
             .withPartFormatFromDisk()
             .build();
@@ -3835,9 +3835,9 @@ MergeTreeDataPartFormat MergeTreeData::choosePartFormatOnDisk(size_t bytes_uncom
 }
 
 MergeTreeDataPartBuilder MergeTreeData::getDataPartBuilder(
-    const String & name, const VolumePtr & volume, const String & part_dir) const
+    const String & name, const VolumePtr & volume, const String & part_dir, const ReadSettings & read_settings_) const
 {
-    return MergeTreeDataPartBuilder(*this, name, volume, relative_data_path, part_dir);
+    return MergeTreeDataPartBuilder(*this, name, volume, relative_data_path, part_dir, read_settings_);
 }
 
 void MergeTreeData::changeSettings(
@@ -5919,7 +5919,7 @@ MergeTreeData::MutableDataPartPtr MergeTreeData::loadPartRestoredFromBackup(cons
     /// Load this part from the directory `temp_part_dir`.
     auto load_part = [&]
     {
-        MergeTreeDataPartBuilder builder(*this, part_name, single_disk_volume, parent_part_dir, part_dir_name);
+        MergeTreeDataPartBuilder builder(*this, part_name, single_disk_volume, parent_part_dir, part_dir_name, getReadSettings());
         builder.withPartFormatFromDisk();
         part = std::move(builder).build();
         part->version.setCreationTID(Tx::PrehistoricTID, nullptr);
@@ -5934,7 +5934,7 @@ MergeTreeData::MutableDataPartPtr MergeTreeData::loadPartRestoredFromBackup(cons
         if (!part)
         {
             /// Make a fake data part only to copy its files to /detached/.
-            part = MergeTreeDataPartBuilder{*this, part_name, single_disk_volume, parent_part_dir, part_dir_name}
+            part = MergeTreeDataPartBuilder{*this, part_name, single_disk_volume, parent_part_dir, part_dir_name, getReadSettings()}
                        .withPartStorageType(MergeTreeDataPartStorageType::Full)
                        .withPartType(MergeTreeDataPartType::Wide)
                        .build();
@@ -6586,7 +6586,7 @@ MergeTreeData::MutableDataPartsVector MergeTreeData::tryLoadPartsToAttach(const
         LOG_DEBUG(log, "Checking part {}", new_name);
 
         auto single_disk_volume = std::make_shared<SingleDiskVolume>("volume_" + old_name, disk);
-        auto part = getDataPartBuilder(old_name, single_disk_volume, source_dir / new_name)
+        auto part = getDataPartBuilder(old_name, single_disk_volume, source_dir / new_name, getReadSettings())
             .withPartFormatFromDisk()
             .build();
 
@@ -7641,7 +7641,7 @@ std::pair<MergeTreeData::MutableDataPartPtr, scope_guard> MergeTreeData::cloneAn
               std::string(fs::path(dst_part_storage->getFullRootPath()) / tmp_dst_part_name),
               with_copy);
 
-    auto dst_data_part = MergeTreeDataPartBuilder(*this, dst_part_name, dst_part_storage)
+    auto dst_data_part = MergeTreeDataPartBuilder(*this, dst_part_name, dst_part_storage, getReadSettings())
         .withPartFormatFromDisk()
         .build();
 
@@ -8910,7 +8910,7 @@ std::pair<MergeTreeData::MutableDataPartPtr, scope_guard> MergeTreeData::createE
     VolumePtr data_part_volume = createVolumeFromReservation(reservation, volume);
 
     auto tmp_dir_holder = getTemporaryPartDirectoryHolder(EMPTY_PART_TMP_PREFIX + new_part_name);
-    auto new_data_part = getDataPartBuilder(new_part_name, data_part_volume, EMPTY_PART_TMP_PREFIX + new_part_name)
+    auto new_data_part = getDataPartBuilder(new_part_name, data_part_volume, EMPTY_PART_TMP_PREFIX + new_part_name, getReadSettings())
         .withBytesAndRowsOnDisk(0, 0)
         .withPartInfo(new_part_info)
         .build();

src/Storages/MergeTree/MergeTreeData.h

@@ -241,7 +241,7 @@ public:
 
     MergeTreeDataPartFormat choosePartFormat(size_t bytes_uncompressed, size_t rows_count) const;
     MergeTreeDataPartFormat choosePartFormatOnDisk(size_t bytes_uncompressed, size_t rows_count) const;
-    MergeTreeDataPartBuilder getDataPartBuilder(const String & name, const VolumePtr & volume, const String & part_dir) const;
+    MergeTreeDataPartBuilder getDataPartBuilder(const String & name, const VolumePtr & volume, const String & part_dir, const ReadSettings & read_settings_) const;
 
     /// Auxiliary object to add a set of parts into the working set in two steps:
     /// * First, as PreActive parts (the parts are ready, but not yet in the active set).

src/Storages/MergeTree/MergeTreeDataPartBuilder.cpp

@@ -14,20 +14,22 @@ namespace ErrorCodes
 }
 
 MergeTreeDataPartBuilder::MergeTreeDataPartBuilder(
-    const MergeTreeData & data_, String name_, VolumePtr volume_, String root_path_, String part_dir_)
+    const MergeTreeData & data_, String name_, VolumePtr volume_, String root_path_, String part_dir_, const ReadSettings & read_settings_)
     : data(data_)
     , name(std::move(name_))
     , volume(std::move(volume_))
     , root_path(std::move(root_path_))
     , part_dir(std::move(part_dir_))
+    , read_settings(read_settings_)
 {
 }
 
 MergeTreeDataPartBuilder::MergeTreeDataPartBuilder(
-    const MergeTreeData & data_, String name_, MutableDataPartStoragePtr part_storage_)
+    const MergeTreeData & data_, String name_, MutableDataPartStoragePtr part_storage_, const ReadSettings & read_settings_)
     : data(data_)
     , name(std::move(name_))
     , part_storage(std::move(part_storage_))
+    , read_settings(read_settings_)
 {
 }
 
@@ -73,7 +75,8 @@ MutableDataPartStoragePtr MergeTreeDataPartBuilder::getPartStorageByType(
     MergeTreeDataPartStorageType storage_type_,
     const VolumePtr & volume_,
     const String & root_path_,
-    const String & part_dir_)
+    const String & part_dir_,
+    const ReadSettings &) /// Unused here, but used in private repo.
 {
     if (!volume_)
         throw Exception(ErrorCodes::LOGICAL_ERROR, "Cannot create part storage, because volume is not specified");
@@ -112,7 +115,7 @@ MergeTreeDataPartBuilder & MergeTreeDataPartBuilder::withPartType(MergeTreeDataP
 
 MergeTreeDataPartBuilder & MergeTreeDataPartBuilder::withPartStorageType(MergeTreeDataPartStorageType storage_type_)
 {
-    part_storage = getPartStorageByType(storage_type_, volume, root_path, part_dir);
+    part_storage = getPartStorageByType(storage_type_, volume, root_path, part_dir, read_settings);
     return *this;
 }
 
@@ -126,7 +129,8 @@ MergeTreeDataPartBuilder::PartStorageAndMarkType
 MergeTreeDataPartBuilder::getPartStorageAndMarkType(
     const VolumePtr & volume_,
     const String & root_path_,
-    const String & part_dir_)
+    const String & part_dir_,
+    const ReadSettings & read_settings_)
 {
     auto disk = volume_->getDisk();
     auto part_relative_path = fs::path(root_path_) / part_dir_;
@@ -138,7 +142,7 @@ MergeTreeDataPartBuilder::getPartStorageAndMarkType(
 
         if (MarkType::isMarkFileExtension(ext))
         {
-            auto storage = getPartStorageByType(MergeTreeDataPartStorageType::Full, volume_, root_path_, part_dir_);
+            auto storage = getPartStorageByType(MergeTreeDataPartStorageType::Full, volume_, root_path_, part_dir_, read_settings_);
             return {std::move(storage), MarkType(ext)};
         }
     }
@@ -156,7 +160,7 @@ MergeTreeDataPartBuilder & MergeTreeDataPartBuilder::withPartFormatFromDisk()
 MergeTreeDataPartBuilder & MergeTreeDataPartBuilder::withPartFormatFromVolume()
 {
     assert(volume);
-    auto [storage, mark_type] = getPartStorageAndMarkType(volume, root_path, part_dir);
+    auto [storage, mark_type] = getPartStorageAndMarkType(volume, root_path, part_dir, read_settings);
 
     if (!storage || !mark_type)
     {

src/Storages/MergeTree/MergeTreeDataPartBuilder.h

@@ -21,8 +21,8 @@ using VolumePtr = std::shared_ptr<IVolume>;
 class MergeTreeDataPartBuilder
 {
 public:
-    MergeTreeDataPartBuilder(const MergeTreeData & data_, String name_, VolumePtr volume_, String root_path_, String part_dir_);
-    MergeTreeDataPartBuilder(const MergeTreeData & data_, String name_, MutableDataPartStoragePtr part_storage_);
+    MergeTreeDataPartBuilder(const MergeTreeData & data_, String name_, VolumePtr volume_, String root_path_, String part_dir_, const ReadSettings & read_settings_);
+    MergeTreeDataPartBuilder(const MergeTreeData & data_, String name_, MutableDataPartStoragePtr part_storage_, const ReadSettings & read_settings_);
 
     std::shared_ptr<IMergeTreeDataPart> build();
 
@@ -42,7 +42,8 @@ public:
     static PartStorageAndMarkType getPartStorageAndMarkType(
         const VolumePtr & volume_,
         const String & root_path_,
-        const String & part_dir_);
+        const String & part_dir_,
+        const ReadSettings & read_settings);
 
 private:
     Self & withPartFormatFromVolume();
@@ -52,7 +53,8 @@ private:
         MergeTreeDataPartStorageType storage_type_,
         const VolumePtr & volume_,
         const String & root_path_,
-        const String & part_dir_);
+        const String & part_dir_,
+        const ReadSettings & read_settings);
 
     const MergeTreeData & data;
     const String name;
@@ -64,6 +66,8 @@ private:
     std::optional<MergeTreeDataPartType> part_type;
     MutableDataPartStoragePtr part_storage;
     const IMergeTreeDataPart * parent_part = nullptr;
+
+    const ReadSettings read_settings;
 };
 
 }

src/Storages/MergeTree/MergeTreeDataWriter.cpp

@@ -610,7 +610,7 @@ MergeTreeDataWriter::TemporaryPart MergeTreeDataWriter::writeTempPartImpl(
         }
     }
 
-    auto new_data_part = data.getDataPartBuilder(part_name, data_part_volume, part_dir)
+    auto new_data_part = data.getDataPartBuilder(part_name, data_part_volume, part_dir, getReadSettings())
         .withPartFormat(data.choosePartFormat(expected_size, block.rows()))
         .withPartInfo(new_part_info)
         .build();

src/Storages/MergeTree/MergeTreePartsMover.cpp

@@ -280,7 +280,7 @@ MergeTreePartsMover::TemporaryClonedPart MergeTreePartsMover::clonePart(const Me
         cloned_part_storage = part->makeCloneOnDisk(disk, MergeTreeData::MOVING_DIR_NAME, read_settings, write_settings, cancellation_hook);
     }
 
-    MergeTreeDataPartBuilder builder(*data, part->name, cloned_part_storage);
+    MergeTreeDataPartBuilder builder(*data, part->name, cloned_part_storage, getReadSettings());
     cloned_part.part = std::move(builder).withPartFormatFromDisk().build();
     LOG_TRACE(log, "Part {} was cloned to {}", part->name, cloned_part.part->getDataPartStorage().getFullPath());
 

src/Storages/MergeTree/MutateTask.cpp

@@ -2289,7 +2289,7 @@ bool MutateTask::prepare()
     String tmp_part_dir_name = prefix + ctx->future_part->name;
     ctx->temporary_directory_lock = ctx->data->getTemporaryPartDirectoryHolder(tmp_part_dir_name);
 
-    auto builder = ctx->data->getDataPartBuilder(ctx->future_part->name, single_disk_volume, tmp_part_dir_name);
+    auto builder = ctx->data->getDataPartBuilder(ctx->future_part->name, single_disk_volume, tmp_part_dir_name, getReadSettings());
     builder.withPartFormat(ctx->future_part->part_format);
     builder.withPartInfo(ctx->future_part->part_info);
 

src/Storages/StorageReplicatedMergeTree.cpp

@@ -2095,7 +2095,7 @@ MergeTreeData::MutableDataPartPtr StorageReplicatedMergeTree::attachPartHelperFo
             const auto part_old_name = part_info->getPartNameV1();
             const auto volume = std::make_shared<SingleDiskVolume>("volume_" + part_old_name, disk);
 
-            auto part = getDataPartBuilder(entry.new_part_name, volume, fs::path(DETACHED_DIR_NAME) / part_old_name)
+            auto part = getDataPartBuilder(entry.new_part_name, volume, fs::path(DETACHED_DIR_NAME) / part_old_name, getReadSettings())
                 .withPartFormatFromDisk()
                 .build();
 

src/Storages/fuzzers/CMakeLists.txt

@@ -4,4 +4,4 @@ clickhouse_add_executable (mergetree_checksum_fuzzer mergetree_checksum_fuzzer.c
 target_link_libraries (mergetree_checksum_fuzzer PRIVATE dbms)
 
 clickhouse_add_executable (columns_description_fuzzer columns_description_fuzzer.cpp)
-target_link_libraries (columns_description_fuzzer PRIVATE)
+target_link_libraries (columns_description_fuzzer PRIVATE dbms)

tests/ci/ci_config.py

@@ -539,9 +539,9 @@ class CI:
         JobNames.LIBFUZZER_TEST: JobConfig(
             required_builds=[BuildNames.FUZZERS],
             run_by_labels=[Tags.libFuzzer],
-            timeout=10800,
+            timeout=5400,
             run_command='libfuzzer_test_check.py "$CHECK_NAME"',
-            runner_type=Runners.STYLE_CHECKER,
+            runner_type=Runners.FUNC_TESTER,
         ),
         JobNames.DOCKER_SERVER: CommonJobConfigs.DOCKER_SERVER.with_properties(
             required_builds=[BuildNames.PACKAGE_RELEASE, BuildNames.PACKAGE_AARCH64]

tests/ci/libfuzzer_test_check.py

@@ -3,20 +3,37 @@
 import argparse
 import logging
 import os
+import re
 import sys
 import zipfile
 from pathlib import Path
 from typing import List
 
+from botocore.exceptions import ClientError
+
 from build_download_helper import download_fuzzers
 from clickhouse_helper import CiLogsCredentials
 from docker_images_helper import DockerImage, get_docker_image, pull_image
-from env_helper import REPO_COPY, REPORT_PATH, TEMP_PATH
+from env_helper import REPO_COPY, REPORT_PATH, S3_BUILDS_BUCKET, TEMP_PATH
 from pr_info import PRInfo
+from report import FAILURE, SUCCESS, JobReport, TestResult
+from s3_helper import S3Helper
 from stopwatch import Stopwatch
 from tee_popen import TeePopen
 
+TIMEOUT = 60
 NO_CHANGES_MSG = "Nothing to run"
+s3 = S3Helper()
+
+
+def zipdir(path, ziph):
+    # ziph is zipfile handle
+    for root, _, files in os.walk(path):
+        for file in files:
+            ziph.write(
+                os.path.join(root, file),
+                os.path.relpath(os.path.join(root, file), os.path.join(path, "..")),
+            )
 
 
 def get_additional_envs(check_name, run_by_hash_num, run_by_hash_total):
@@ -59,16 +76,19 @@ def get_run_command(
 
     envs = [
         # a static link, don't use S3_URL or S3_DOWNLOAD
-        '-e S3_URL="https://s3.amazonaws.com/clickhouse-datasets"',
+        '-e S3_URL="https://s3.amazonaws.com"',
     ]
 
     envs += [f"-e {e}" for e in additional_envs]
 
     env_str = " ".join(envs)
+    uid = os.getuid()
+    gid = os.getgid()
 
     return (
         f"docker run "
         f"{ci_logs_args} "
+        f"--user {uid}:{gid} "
         f"--workdir=/fuzzers "
         f"--volume={fuzzers_path}:/fuzzers "
         f"--volume={repo_path}/tests:/usr/share/clickhouse-test "
@@ -85,6 +105,115 @@ def parse_args():
     return parser.parse_args()
 
 
+def download_corpus(path: str):
+    logging.info("Download corpus...")
+
+    try:
+        s3.download_file(
+            bucket=S3_BUILDS_BUCKET,
+            s3_path="fuzzer/corpus.zip",
+            local_file_path=path,
+        )
+    except ClientError as e:
+        if e.response["Error"]["Code"] == "NoSuchKey":
+            logging.debug("No active corpus exists")
+        else:
+            raise
+
+    with zipfile.ZipFile(f"{path}/corpus.zip", "r") as zipf:
+        zipf.extractall(path)
+    os.remove(f"{path}/corpus.zip")
+
+    units = 0
+    for _, _, files in os.walk(path):
+        units += len(files)
+
+    logging.info("...downloaded %d units", units)
+
+
+def upload_corpus(path: str):
+    with zipfile.ZipFile(f"{path}/corpus.zip", "w", zipfile.ZIP_DEFLATED) as zipf:
+        zipdir(f"{path}/corpus/", zipf)
+    s3.upload_file(
+        bucket=S3_BUILDS_BUCKET,
+        file_path=f"{path}/corpus.zip",
+        s3_path="fuzzer/corpus.zip",
+    )
+
+
+def process_error(path: Path) -> list:
+    ERROR = r"^==\d+==\s?ERROR: (\S+): (.*)"
+    # error_source = ""
+    # error_reason = ""
+    # test_unit = ""
+    # TEST_UNIT_LINE = r"artifact_prefix='.*\/'; Test unit written to (.*)"
+    error_info = []
+    is_error = False
+
+    with open(path, "r", encoding="utf-8") as file:
+        for line in file:
+            line = line.rstrip("\n")
+            if is_error:
+                error_info.append(line)
+                # match = re.search(TEST_UNIT_LINE, line)
+                # if match:
+                #     test_unit = match.group(1)
+                continue
+
+            match = re.search(ERROR, line)
+            if match:
+                error_info.append(line)
+                # error_source = match.group(1)
+                # error_reason = match.group(2)
+                is_error = True
+
+    return error_info
+
+
+def read_status(status_path: Path):
+    result = []
+    with open(status_path, "r", encoding="utf-8") as file:
+        for line in file:
+            result.append(line.rstrip("\n"))
+    return result
+
+
+def process_results(result_path: Path):
+    test_results = []
+    oks = 0
+    errors = 0
+    fails = 0
+    for file in result_path.glob("*.status"):
+        fuzzer = file.stem
+        file_path = file.parent / fuzzer
+        file_path_unit = file_path.with_suffix(".unit")
+        file_path_out = file_path.with_suffix(".out")
+        file_path_stdout = file_path.with_suffix(".stdout")
+        status = read_status(file)
+        result = TestResult(fuzzer, status[0], float(status[2]))
+        if status[0] == "OK":
+            oks += 1
+        elif status[0] == "ERROR":
+            errors += 1
+            if file_path_out.exists():
+                result.set_log_files(f"['{file_path_out}']")
+            elif file_path_stdout.exists():
+                result.set_log_files(f"['{file_path_stdout}']")
+        else:
+            fails += 1
+            if file_path_out.exists():
+                result.set_raw_logs("\n".join(process_error(file_path_out)))
+            if file_path_unit.exists():
+                result.set_log_files(f"['{file_path_unit}']")
+            elif file_path_out.exists():
+                result.set_log_files(f"['{file_path_out}']")
+            elif file_path_stdout.exists():
+                result.set_log_files(f"['{file_path_stdout}']")
+        test_results.append(result)
+
+    return [oks, errors, fails, test_results]
+
+
 def main():
     logging.basicConfig(level=logging.INFO)
 
@@ -114,15 +243,18 @@ def main():
     fuzzers_path = temp_path / "fuzzers"
     fuzzers_path.mkdir(parents=True, exist_ok=True)
 
+    download_corpus(fuzzers_path)
     download_fuzzers(check_name, reports_path, fuzzers_path)
 
     for file in os.listdir(fuzzers_path):
         if file.endswith("_fuzzer"):
             os.chmod(fuzzers_path / file, 0o777)
         elif file.endswith("_seed_corpus.zip"):
-            corpus_path = fuzzers_path / (file.removesuffix("_seed_corpus.zip") + ".in")
+            seed_corpus_path = fuzzers_path / (
+                file.removesuffix("_seed_corpus.zip") + ".in"
+            )
             with zipfile.ZipFile(fuzzers_path / file, "r") as zfd:
-                zfd.extractall(corpus_path)
+                zfd.extractall(seed_corpus_path)
 
     result_path = temp_path / "result_path"
     result_path.mkdir(parents=True, exist_ok=True)
@@ -133,6 +265,8 @@ def main():
         check_name, run_by_hash_num, run_by_hash_total
     )
 
+    additional_envs.append(f"TIMEOUT={TIMEOUT}")
+
     ci_logs_credentials = CiLogsCredentials(Path(temp_path) / "export-logs-config.sh")
     ci_logs_args = ci_logs_credentials.get_docker_arguments(
         pr_info, stopwatch.start_time_str, check_name
@@ -152,10 +286,25 @@ def main():
         retcode = process.wait()
         if retcode == 0:
             logging.info("Run successfully")
+            upload_corpus(fuzzers_path)
         else:
             logging.info("Run failed")
 
-    sys.exit(0)
+    results = process_results(result_path)
+
+    success = results[1] == 0 and results[2] == 0
+
+    JobReport(
+        description=f"OK: {results[0]}, ERROR: {results[1]}, FAIL: {results[2]}",
+        test_results=results[3],
+        status=SUCCESS if success else FAILURE,
+        start_time=stopwatch.start_time_str,
+        duration=stopwatch.duration_seconds,
+        additional_files=[],
+    ).dump()
+
+    if not success:
+        sys.exit(1)
 
 
 if __name__ == "__main__":

tests/ci/mark_release_ready.py

@@ -9,9 +9,10 @@ from get_robot_token import get_best_robot_token
 from git_helper import commit as commit_arg
 from github_helper import GitHub
 from pr_info import PRInfo
-from release import RELEASE_READY_STATUS
 from report import SUCCESS
 
+RELEASE_READY_STATUS = "Ready for release"
+
 
 def main():
     parser = argparse.ArgumentParser(

tests/ci/release.py

@@ -1,693 +0,0 @@
-#!/usr/bin/env python3
-
-"""
-script to create releases for ClickHouse
-
-The `gh` CLI preferred over the PyGithub to have an easy way to rollback bad
-release in command line by simple execution giving rollback commands
-
-On another hand, PyGithub is used for convenient getting commit's status from API
-
-To run this script on a freshly installed Ubuntu 22.04 system, it is enough to do the following commands:
-
-sudo apt install pip
-pip install requests boto3 github PyGithub
-sudo snap install gh
-gh auth login
-"""
-
-
-import argparse
-import json
-import logging
-import subprocess
-from contextlib import contextmanager
-from typing import Any, Final, Iterator, List, Optional, Tuple
-
-from ci_config import Labels
-from git_helper import Git, commit, release_branch
-from report import SUCCESS
-from version_helper import (
-    FILE_WITH_VERSION_PATH,
-    GENERATED_CONTRIBUTORS,
-    ClickHouseVersion,
-    VersionType,
-    get_abs_path,
-    get_version_from_repo,
-    update_cmake_version,
-    update_contributors,
-)
-
-RELEASE_READY_STATUS = "Ready for release"
-
-
-class Repo:
-    VALID = ("ssh", "https", "origin")
-
-    def __init__(self, repo: str, protocol: str):
-        self._repo = repo
-        self._url = ""
-        self.url = protocol
-
-    @property
-    def url(self) -> str:
-        return self._url
-
-    @url.setter
-    def url(self, protocol: str) -> None:
-        if protocol == "ssh":
-            self._url = f"git@github.com:{self}.git"
-        elif protocol == "https":
-            self._url = f"https://github.com/{self}.git"
-        elif protocol == "origin":
-            self._url = protocol
-        else:
-            raise ValueError(f"protocol must be in {self.VALID}")
-
-    def __str__(self):
-        return self._repo
-
-
-class Release:
-    NEW = "new"  # type: Final
-    PATCH = "patch"  # type: Final
-    VALID_TYPE = (NEW, PATCH)  # type: Final[Tuple[str, str]]
-    CMAKE_PATH = get_abs_path(FILE_WITH_VERSION_PATH)
-    CONTRIBUTORS_PATH = get_abs_path(GENERATED_CONTRIBUTORS)
-
-    def __init__(
-        self,
-        repo: Repo,
-        release_commit: str,
-        release_type: str,
-        dry_run: bool,
-        with_stderr: bool,
-    ):
-        self.repo = repo
-        self._release_commit = ""
-        self.release_commit = release_commit
-        self.dry_run = dry_run
-        self.with_stderr = with_stderr
-        assert release_type in self.VALID_TYPE
-        self.release_type = release_type
-        self._git = Git()
-        self._version = get_version_from_repo(git=self._git)
-        self.release_version = self.version
-        self._release_branch = ""
-        self._version_new_tag = None  # type: Optional[ClickHouseVersion]
-        self._rollback_stack = []  # type: List[str]
-
-    def run(
-        self, cmd: str, cwd: Optional[str] = None, dry_run: bool = False, **kwargs: Any
-    ) -> str:
-        cwd_text = ""
-        if cwd:
-            cwd_text = f" (CWD='{cwd}')"
-        if dry_run:
-            logging.info("Would run command%s:\n    %s", cwd_text, cmd)
-            return ""
-        if not self.with_stderr:
-            kwargs["stderr"] = subprocess.DEVNULL
-
-        logging.info("Running command%s:\n    %s", cwd_text, cmd)
-        return self._git.run(cmd, cwd, **kwargs)
-
-    def set_release_info(self):
-        # Fetch release commit and tags in case they don't exist locally
-        self.run(
-            f"git fetch {self.repo.url} {self.release_commit} --no-recurse-submodules"
-        )
-        self.run(f"git fetch {self.repo.url} --tags --no-recurse-submodules")
-
-        # Get the actual version for the commit before check
-        with self._checkout(self.release_commit, True):
-            self.release_branch = f"{self.version.major}.{self.version.minor}"
-            self.release_version = get_version_from_repo(git=self._git)
-            self.release_version.with_description(self.get_stable_release_type())
-
-        self.read_version()
-
-    def read_version(self):
-        self._git.update()
-        self.version = get_version_from_repo(git=self._git)
-
-    def get_stable_release_type(self) -> str:
-        if self.version.is_lts:
-            return VersionType.LTS
-        return VersionType.STABLE
-
-    def check_commit_release_ready(self):
-        per_page = 100
-        page = 1
-        while True:
-            statuses = json.loads(
-                self.run(
-                    f"gh api 'repos/{self.repo}/commits/{self.release_commit}"
-                    f"/statuses?per_page={per_page}&page={page}'"
-                )
-            )
-
-            if not statuses:
-                break
-
-            for status in statuses:
-                if status["context"] == RELEASE_READY_STATUS:
-                    if not status["state"] == SUCCESS:
-                        raise ValueError(
-                            f"the status {RELEASE_READY_STATUS} is {status['state']}"
-                            ", not success"
-                        )
-
-                    return
-
-            page += 1
-
-        raise KeyError(
-            f"the status {RELEASE_READY_STATUS} "
-            f"is not found for commit {self.release_commit}"
-        )
-
-    def check_prerequisites(self):
-        """
-        Check tooling installed in the system, `git` is checked by Git() init
-        """
-        try:
-            self.run("gh auth status")
-        except subprocess.SubprocessError:
-            logging.error(
-                "The github-cli either not installed or not setup, please follow "
-                "the instructions on https://github.com/cli/cli#installation and "
-                "https://cli.github.com/manual/"
-            )
-            raise
-
-        if self.release_type == self.PATCH:
-            self.check_commit_release_ready()
-
-    def do(
-        self, check_dirty: bool, check_run_from_master: bool, check_branch: bool
-    ) -> None:
-        self.check_prerequisites()
-
-        if check_dirty:
-            logging.info("Checking if repo is clean")
-            try:
-                self.run("git diff HEAD --exit-code")
-            except subprocess.CalledProcessError:
-                logging.fatal("Repo contains uncommitted changes")
-                raise
-
-        if check_run_from_master and self._git.branch != "master":
-            raise RuntimeError("the script must be launched only from master")
-
-        self.set_release_info()
-
-        if check_branch:
-            self.check_branch()
-
-        if self.release_type == self.NEW:
-            with self._checkout(self.release_commit, True):
-                # Checkout to the commit, it will provide the correct current version
-                with self.new_release():
-                    with self.create_release_branch():
-                        logging.info(
-                            "Publishing release %s from commit %s is done",
-                            self.release_version.describe,
-                            self.release_commit,
-                        )
-
-        elif self.release_type == self.PATCH:
-            with self._checkout(self.release_commit, True):
-                with self.patch_release():
-                    logging.info(
-                        "Publishing release %s from commit %s is done",
-                        self.release_version.describe,
-                        self.release_commit,
-                    )
-
-        if self.dry_run:
-            logging.info("Dry running, clean out possible changes")
-            rollback = self._rollback_stack.copy()
-            rollback.reverse()
-            for cmd in rollback:
-                self.run(cmd)
-            return
-
-        self.log_post_workflows()
-        self.log_rollback()
-
-    def check_no_tags_after(self):
-        tags_after_commit = self.run(f"git tag --contains={self.release_commit}")
-        if tags_after_commit:
-            raise RuntimeError(
-                f"Commit {self.release_commit} belongs to following tags:\n"
-                f"{tags_after_commit}\nChoose another commit"
-            )
-
-    def check_branch(self):
-        branch = self.release_branch
-        if self.release_type == self.NEW:
-            # Commit to spin up the release must belong to a main branch
-            branch = "master"
-        elif self.release_type != self.PATCH:
-            raise (
-                ValueError(f"release_type {self.release_type} not in {self.VALID_TYPE}")
-            )
-
-        # Prefetch the branch to have it updated
-        if self._git.branch == branch:
-            self.run("git pull --no-recurse-submodules")
-        else:
-            self.run(
-                f"git fetch {self.repo.url} {branch}:{branch} --no-recurse-submodules"
-            )
-        output = self.run(f"git branch --contains={self.release_commit} {branch}")
-        if branch not in output:
-            raise RuntimeError(
-                f"commit {self.release_commit} must belong to {branch} "
-                f"for {self.release_type} release"
-            )
-
-    def _update_cmake_contributors(
-        self, version: ClickHouseVersion, reset_tweak: bool = True
-    ) -> None:
-        if reset_tweak:
-            desc = version.description
-            version = version.reset_tweak()
-            version.with_description(desc)
-        update_cmake_version(version)
-        update_contributors(raise_error=True)
-        if self.dry_run:
-            logging.info(
-                "Dry running, resetting the following changes in the repo:\n%s",
-                self.run(f"git diff '{self.CMAKE_PATH}' '{self.CONTRIBUTORS_PATH}'"),
-            )
-            self.run(f"git checkout '{self.CMAKE_PATH}' '{self.CONTRIBUTORS_PATH}'")
-
-    def _commit_cmake_contributors(
-        self, version: ClickHouseVersion, reset_tweak: bool = True
-    ) -> None:
-        if reset_tweak:
-            version = version.reset_tweak()
-        self.run(
-            f"git commit '{self.CMAKE_PATH}' '{self.CONTRIBUTORS_PATH}' "
-            f"-m 'Update autogenerated version to {version.string} and contributors'",
-            dry_run=self.dry_run,
-        )
-
-    @property
-    def bump_part(self) -> ClickHouseVersion.PART_TYPE:
-        if self.release_type == Release.NEW:
-            if self._version.minor >= 12:
-                return "major"
-            return "minor"
-        return "patch"
-
-    @property
-    def has_rollback(self) -> bool:
-        return bool(self._rollback_stack)
-
-    def log_rollback(self):
-        if self.has_rollback:
-            rollback = self._rollback_stack.copy()
-            rollback.reverse()
-            logging.info(
-                "To rollback the action run the following commands:\n  %s",
-                "\n  ".join(rollback),
-            )
-
-    def log_post_workflows(self):
-        logging.info(
-            "To verify all actions are running good visit the following links:\n  %s",
-            "\n  ".join(
-                f"https://github.com/{self.repo}/actions/workflows/{action}.yml"
-                for action in ("release", "tags_stable")
-            ),
-        )
-
-    @contextmanager
-    def create_release_branch(self):
-        self.check_no_tags_after()
-        # Create release branch
-        self.read_version()
-        assert self._version_new_tag is not None
-        with self._create_tag(
-            self._version_new_tag.describe,
-            self.release_commit,
-            f"Initial commit for release {self._version_new_tag.major}.{self._version_new_tag.minor}",
-        ):
-            with self._create_branch(self.release_branch, self.release_commit):
-                with self._checkout(self.release_branch, True):
-                    with self._bump_release_branch():
-                        yield
-
-    @contextmanager
-    def patch_release(self):
-        self.check_no_tags_after()
-        self.read_version()
-        version_type = self.get_stable_release_type()
-        self.version.with_description(version_type)
-        with self._create_gh_release(False):
-            self.version = self.version.update(self.bump_part)
-            self.version.with_description(version_type)
-            self._update_cmake_contributors(self.version)
-            # Checking out the commit of the branch and not the branch itself,
-            # then we are able to skip rollback
-            with self._checkout(f"{self.release_branch}^0", False):
-                current_commit = self.run("git rev-parse HEAD")
-                self._commit_cmake_contributors(self.version)
-                with self._push(
-                    "HEAD", with_rollback_on_fail=False, remote_ref=self.release_branch
-                ):
-                    # DO NOT PUT ANYTHING ELSE HERE
-                    # The push must be the last action and mean the successful release
-                    self._rollback_stack.append(
-                        f"{self.dry_run_prefix}git push {self.repo.url} "
-                        f"+{current_commit}:{self.release_branch}"
-                    )
-                    yield
-
-    @contextmanager
-    def new_release(self):
-        # Create branch for a version bump
-        self.read_version()
-        self.version = self.version.update(self.bump_part)
-        helper_branch = f"{self.version.major}.{self.version.minor}-prepare"
-        with self._create_branch(helper_branch, self.release_commit):
-            with self._checkout(helper_branch, True):
-                with self._bump_version_in_master(helper_branch):
-                    yield
-
-    @property
-    def version(self) -> ClickHouseVersion:
-        return self._version
-
-    @version.setter
-    def version(self, version: ClickHouseVersion) -> None:
-        if not isinstance(version, ClickHouseVersion):
-            raise ValueError(f"version must be ClickHouseVersion, not {type(version)}")
-        self._version = version
-
-    @property
-    def release_branch(self) -> str:
-        return self._release_branch
-
-    @release_branch.setter
-    def release_branch(self, branch: str) -> None:
-        self._release_branch = release_branch(branch)
-
-    @property
-    def release_commit(self) -> str:
-        return self._release_commit
-
-    @release_commit.setter
-    def release_commit(self, release_commit: str) -> None:
-        self._release_commit = commit(release_commit)
-
-    @property
-    def dry_run_prefix(self) -> str:
-        if self.dry_run:
-            return "# "
-        return ""
-
-    @contextmanager
-    def _bump_release_branch(self):
-        # Update only git, original version stays the same
-        self._git.update()
-        new_version = self.version.copy()
-        version_type = self.get_stable_release_type()
-        pr_labels = f"--label {Labels.RELEASE}"
-        if version_type == VersionType.LTS:
-            pr_labels += f" --label {Labels.RELEASE_LTS}"
-        new_version.with_description(version_type)
-        self._update_cmake_contributors(new_version)
-        self._commit_cmake_contributors(new_version)
-        with self._push(self.release_branch):
-            with self._create_gh_label(
-                f"v{self.release_branch}-must-backport", "10dbed"
-            ):
-                with self._create_gh_label(
-                    f"v{self.release_branch}-affected", "c2bfff"
-                ):
-                    # The following command is rolled back by deleting branch
-                    # in self._push
-                    self.run(
-                        f"gh pr create --repo {self.repo} --title "
-                        f"'Release pull request for branch {self.release_branch}' "
-                        f"--head {self.release_branch} {pr_labels} "
-                        "--body 'This PullRequest is a part of ClickHouse release "
-                        "cycle. It is used by CI system only. Do not perform any "
-                        "changes with it.'",
-                        dry_run=self.dry_run,
-                    )
-                    # Here the release branch part is done.
-                    # We don't create a release itself automatically to have a
-                    # safe window to backport possible bug fixes.
-                    yield
-
-    @contextmanager
-    def _bump_version_in_master(self, helper_branch: str) -> Iterator[None]:
-        self.read_version()
-        self.version = self.version.update(self.bump_part)
-        self.version.with_description(VersionType.TESTING)
-        self._update_cmake_contributors(self.version)
-        self._commit_cmake_contributors(self.version)
-        # Create a version-new tag
-        self._version_new_tag = self.version.copy()
-        self._version_new_tag.tweak = 1
-        self._version_new_tag.with_description(VersionType.NEW)
-
-        with self._push(helper_branch):
-            body_file = get_abs_path(".github/PULL_REQUEST_TEMPLATE.md")
-            # The following command is rolled back by deleting branch in self._push
-            self.run(
-                f"gh pr create --repo {self.repo} --title 'Update version after "
-                f"release' --head {helper_branch} --body-file '{body_file}' "
-                "--label 'do not test' --assignee @me",
-                dry_run=self.dry_run,
-            )
-            # Here the new release part is done
-            yield
-
-    @contextmanager
-    def _checkout(self, ref: str, with_checkout_back: bool = False) -> Iterator[None]:
-        self._git.update()
-        orig_ref = self._git.branch or self._git.sha
-        rollback_cmd = ""
-        if ref not in (self._git.branch, self._git.sha):
-            self.run(f"git checkout {ref}")
-            # checkout is not put into rollback_stack intentionally
-            rollback_cmd = f"git checkout {orig_ref}"
-            # always update version and git after checked out ref
-            self.read_version()
-        try:
-            yield
-        except (Exception, KeyboardInterrupt):
-            logging.warning("Rolling back checked out %s for %s", ref, orig_ref)
-            self.run(f"git reset --hard; git checkout -f {orig_ref}")
-            raise
-        # Normal flow when we need to checkout back
-        if with_checkout_back and rollback_cmd:
-            self.run(rollback_cmd)
-
-    @contextmanager
-    def _create_branch(self, name: str, start_point: str = "") -> Iterator[None]:
-        self.run(f"git branch {name} {start_point}")
-
-        rollback_cmd = f"git branch -D {name}"
-        self._rollback_stack.append(rollback_cmd)
-        try:
-            yield
-        except (Exception, KeyboardInterrupt):
-            logging.warning("Rolling back created branch %s", name)
-            self.run(rollback_cmd)
-            raise
-
-    @contextmanager
-    def _create_gh_label(self, label: str, color_hex: str) -> Iterator[None]:
-        # API call, https://docs.github.com/en/rest/reference/issues#create-a-label
-        self.run(
-            f"gh api repos/{self.repo}/labels -f name={label} -f color={color_hex}",
-            dry_run=self.dry_run,
-        )
-        rollback_cmd = (
-            f"{self.dry_run_prefix}gh api repos/{self.repo}/labels/{label} -X DELETE"
-        )
-        self._rollback_stack.append(rollback_cmd)
-        try:
-            yield
-        except (Exception, KeyboardInterrupt):
-            logging.warning("Rolling back label %s", label)
-            self.run(rollback_cmd)
-            raise
-
-    @contextmanager
-    def _create_gh_release(self, as_prerelease: bool) -> Iterator[None]:
-        tag = self.release_version.describe
-        with self._create_tag(tag, self.release_commit):
-            # Preserve tag if version is changed
-            prerelease = ""
-            if as_prerelease:
-                prerelease = "--prerelease"
-            self.run(
-                f"gh release create {prerelease} --repo {self.repo} "
-                f"--title 'Release {tag}' '{tag}'",
-                dry_run=self.dry_run,
-            )
-            rollback_cmd = (
-                f"{self.dry_run_prefix}gh release delete --yes "
-                f"--repo {self.repo} '{tag}'"
-            )
-            self._rollback_stack.append(rollback_cmd)
-            try:
-                yield
-            except (Exception, KeyboardInterrupt):
-                logging.warning("Rolling back release publishing")
-                self.run(rollback_cmd)
-                raise
-
-    @contextmanager
-    def _create_tag(
-        self, tag: str, commit: str, tag_message: str = ""
-    ) -> Iterator[None]:
-        tag_message = tag_message or f"Release {tag}"
-        # Create tag even in dry-run
-        self.run(f"git tag -a -m '{tag_message}' '{tag}' {commit}")
-        rollback_cmd = f"git tag -d '{tag}'"
-        self._rollback_stack.append(rollback_cmd)
-        try:
-            with self._push(tag):
-                yield
-        except (Exception, KeyboardInterrupt):
-            logging.warning("Rolling back tag %s", tag)
-            self.run(rollback_cmd)
-            raise
-
-    @contextmanager
-    def _push(
-        self, ref: str, with_rollback_on_fail: bool = True, remote_ref: str = ""
-    ) -> Iterator[None]:
-        if remote_ref == "":
-            remote_ref = ref
-
-        self.run(f"git push {self.repo.url} {ref}:{remote_ref}", dry_run=self.dry_run)
-        if with_rollback_on_fail:
-            rollback_cmd = (
-                f"{self.dry_run_prefix}git push -d {self.repo.url} {remote_ref}"
-            )
-            self._rollback_stack.append(rollback_cmd)
-
-        try:
-            yield
-        except (Exception, KeyboardInterrupt):
-            if with_rollback_on_fail:
-                logging.warning("Rolling back pushed ref %s", ref)
-                self.run(rollback_cmd)
-
-            raise
-
-
-def parse_args() -> argparse.Namespace:
-    parser = argparse.ArgumentParser(
-        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
-        description="Script to release a new ClickHouse version, requires `git` and "
-        "`gh` (github-cli) commands "
-        "!!! LAUNCH IT ONLY FROM THE MASTER BRANCH !!!",
-    )
-
-    parser.add_argument(
-        "--commit",
-        required=True,
-        type=commit,
-        help="commit create a release",
-    )
-    parser.add_argument(
-        "--repo",
-        default="ClickHouse/ClickHouse",
-        help="repository to create the release",
-    )
-    parser.add_argument(
-        "--remote-protocol",
-        "-p",
-        default="ssh",
-        choices=Repo.VALID,
-        help="repo protocol for git commands remote, 'origin' is a special case and "
-        "uses 'origin' as a remote",
-    )
-    parser.add_argument(
-        "--type",
-        required=True,
-        choices=Release.VALID_TYPE,
-        dest="release_type",
-        help="a release type to bump the major.minor.patch version part, "
-        "new branch is created only for the value 'new'",
-    )
-    parser.add_argument("--with-release-branch", default=True, help=argparse.SUPPRESS)
-    parser.add_argument("--check-dirty", default=True, help=argparse.SUPPRESS)
-    parser.add_argument(
-        "--no-check-dirty",
-        dest="check_dirty",
-        action="store_false",
-        default=argparse.SUPPRESS,
-        help="(dangerous) if set, skip check repository for uncommitted changes",
-    )
-    parser.add_argument("--check-run-from-master", default=True, help=argparse.SUPPRESS)
-    parser.add_argument(
-        "--no-run-from-master",
-        dest="check_run_from_master",
-        action="store_false",
-        default=argparse.SUPPRESS,
-        help="(for development) if set, the script could run from non-master branch",
-    )
-    parser.add_argument("--check-branch", default=True, help=argparse.SUPPRESS)
-    parser.add_argument(
-        "--no-check-branch",
-        dest="check_branch",
-        action="store_false",
-        default=argparse.SUPPRESS,
-        help="(debug or development only, dangerous) if set, skip the branch check for "
-        "a run. By default, 'new' type work only for master, and 'patch' "
-        "works only for a release branches, that name "
-        "should be the same as '$MAJOR.$MINOR' version, e.g. 22.2",
-    )
-    parser.add_argument(
-        "--dry-run",
-        action="store_true",
-        help="do not make any actual changes in the repo, just show what will be done",
-    )
-    parser.add_argument(
-        "--with-stderr",
-        action="store_true",
-        help="if set, the stderr of all subprocess commands will be printed as well",
-    )
-
-    return parser.parse_args()
-
-
-def main():
-    logging.basicConfig(level=logging.INFO)
-    args = parse_args()
-    repo = Repo(args.repo, args.remote_protocol)
-    release = Release(
-        repo, args.commit, args.release_type, args.dry_run, args.with_stderr
-    )
-
-    try:
-        release.do(args.check_dirty, args.check_run_from_master, args.check_branch)
-    except:
-        if release.has_rollback:
-            logging.error(
-                "!!The release process finished with error, read the output carefully!!"
-            )
-            logging.error(
-                "Probably, rollback finished with error. "
-                "If you don't see any of the following commands in the output, "
-                "execute them manually:"
-            )
-            release.log_rollback()
-        raise
-
-
-if __name__ == "__main__":
-    assert False, "Script Deprecated, ask ci team for help"
-    main()

tests/ci/s3_helper.py

@@ -311,24 +311,32 @@ class S3Helper:
     def list_prefix(
         self, s3_prefix_path: str, bucket: str = S3_BUILDS_BUCKET
     ) -> List[str]:
-        objects = self.client.list_objects_v2(Bucket=bucket, Prefix=s3_prefix_path)
+        paginator = self.client.get_paginator("list_objects_v2")
+        pages = paginator.paginate(Bucket=bucket, Prefix=s3_prefix_path)
         result = []
-        if "Contents" in objects:
-            for obj in objects["Contents"]:
-                result.append(obj["Key"])
+        for page in pages:
+            if "Contents" in page:
+                for obj in page["Contents"]:
+                    result.append(obj["Key"])
 
         return result
 
     def list_prefix_non_recursive(
-        self, s3_prefix_path: str, bucket: str = S3_BUILDS_BUCKET
+        self,
+        s3_prefix_path: str,
+        bucket: str = S3_BUILDS_BUCKET,
+        only_dirs: bool = False,
     ) -> List[str]:
-        objects = self.client.list_objects_v2(Bucket=bucket, Prefix=s3_prefix_path)
+        paginator = self.client.get_paginator("list_objects_v2")
+        pages = paginator.paginate(Bucket=bucket, Prefix=s3_prefix_path, Delimiter="/")
         result = []
-        if "Contents" in objects:
-            for obj in objects["Contents"]:
-                if "/" not in obj["Key"][len(s3_prefix_path) + 1 :]:
+        for page in pages:
+            if not only_dirs and "Contents" in page:
+                for obj in page["Contents"]:
                     result.append(obj["Key"])
-
+            if "CommonPrefixes" in page:
+                for obj in page["CommonPrefixes"]:
+                    result.append(obj["Prefix"])
         return result
 
     def url_if_exists(self, key: str, bucket: str = S3_BUILDS_BUCKET) -> str:

tests/fuzz/build.sh

@@ -1,5 +1,8 @@
 #!/bin/bash -eu
 
+# rename clickhouse
+mv $OUT/clickhouse $OUT/clickhouse_fuzzer
+
 # copy fuzzer options and dictionaries
 cp $SRC/tests/fuzz/*.dict $OUT/
 cp $SRC/tests/fuzz/*.options $OUT/

tests/fuzz/clickhouse_fuzzer.options

@@ -0,0 +1,2 @@
+[CI]
+FUZZER_ARGS = true

tests/fuzz/runner.py

@@ -1,26 +1,49 @@
 #!/usr/bin/env python3
 
 import configparser
+import datetime
 import logging
 import os
 import subprocess
 from pathlib import Path
 
 DEBUGGER = os.getenv("DEBUGGER", "")
-FUZZER_ARGS = os.getenv("FUZZER_ARGS", "")
+TIMEOUT = int(os.getenv("TIMEOUT", "0"))
+OUTPUT = "/test_output"
 
 
-def run_fuzzer(fuzzer: str):
+class Stopwatch:
+    def __init__(self):
+        self.reset()
+
+    @property
+    def duration_seconds(self) -> float:
+        return (datetime.datetime.utcnow() - self.start_time).total_seconds()
+
+    @property
+    def start_time_str(self) -> str:
+        return self.start_time_str_value
+
+    def reset(self) -> None:
+        self.start_time = datetime.datetime.utcnow()
+        self.start_time_str_value = self.start_time.strftime("%Y-%m-%d %H:%M:%S")
+
+
+def run_fuzzer(fuzzer: str, timeout: int):
     logging.info("Running fuzzer %s...", fuzzer)
 
-    corpus_dir = f"{fuzzer}.in"
-    with Path(corpus_dir) as path:
+    seed_corpus_dir = f"{fuzzer}.in"
+    with Path(seed_corpus_dir) as path:
         if not path.exists() or not path.is_dir():
-            corpus_dir = ""
+            seed_corpus_dir = ""
 
+    active_corpus_dir = f"corpus/{fuzzer}"
+    if not os.path.exists(active_corpus_dir):
+        os.makedirs(active_corpus_dir)
     options_file = f"{fuzzer}.options"
     custom_libfuzzer_options = ""
     fuzzer_arguments = ""
+    use_fuzzer_args = False
 
     with Path(options_file) as path:
         if path.exists() and path.is_file():
@@ -44,7 +67,9 @@ def run_fuzzer(fuzzer: str):
 
             if parser.has_section("libfuzzer"):
                 custom_libfuzzer_options = " ".join(
-                    f"-{key}={value}" for key, value in parser["libfuzzer"].items()
+                    f"-{key}={value}"
+                    for key, value in parser["libfuzzer"].items()
+                    if key not in ("jobs", "exact_artifact_path")
                 )
 
             if parser.has_section("fuzzer_arguments"):
@@ -53,19 +78,70 @@ def run_fuzzer(fuzzer: str):
                     for key, value in parser["fuzzer_arguments"].items()
                 )
 
-    cmd_line = f"{DEBUGGER} ./{fuzzer} {FUZZER_ARGS} {corpus_dir}"
-    if custom_libfuzzer_options:
-        cmd_line += f" {custom_libfuzzer_options}"
-    if fuzzer_arguments:
-        cmd_line += f" {fuzzer_arguments}"
+            use_fuzzer_args = parser.getboolean("CI", "FUZZER_ARGS", fallback=False)
 
-    if not "-dict=" in cmd_line and Path(f"{fuzzer}.dict").exists():
-        cmd_line += f" -dict={fuzzer}.dict"
+    exact_artifact_path = f"{OUTPUT}/{fuzzer}.unit"
+    status_path = f"{OUTPUT}/{fuzzer}.status"
+    out_path = f"{OUTPUT}/{fuzzer}.out"
+    stdout_path = f"{OUTPUT}/{fuzzer}.stdout"
 
-    cmd_line += " < /dev/null"
+    if not "-dict=" in custom_libfuzzer_options and Path(f"{fuzzer}.dict").exists():
+        custom_libfuzzer_options += f" -dict={fuzzer}.dict"
+    custom_libfuzzer_options += f" -exact_artifact_path={exact_artifact_path}"
 
-    logging.info("...will execute: %s", cmd_line)
-    subprocess.check_call(cmd_line, shell=True)
+    libfuzzer_corpora = f"{active_corpus_dir} {seed_corpus_dir}"
+
+    cmd_line = f"{DEBUGGER} ./{fuzzer} {fuzzer_arguments}"
+
+    env = None
+    with_fuzzer_args = ""
+    if use_fuzzer_args:
+        env = {"FUZZER_ARGS": f"{custom_libfuzzer_options} {libfuzzer_corpora}".strip()}
+        with_fuzzer_args = f" with FUZZER_ARGS '{env['FUZZER_ARGS']}'"
+    else:
+        cmd_line += f" {custom_libfuzzer_options} {libfuzzer_corpora}"
+
+    logging.info("...will execute: '%s'%s", cmd_line, with_fuzzer_args)
+
+    stopwatch = Stopwatch()
+    try:
+        with open(out_path, "wb") as out, open(stdout_path, "wb") as stdout:
+            subprocess.run(
+                cmd_line.split(),
+                stdin=subprocess.DEVNULL,
+                stdout=stdout,
+                stderr=out,
+                text=True,
+                check=True,
+                shell=False,
+                errors="replace",
+                timeout=timeout,
+                env=env,
+            )
+    except subprocess.CalledProcessError:
+        logging.info("Fail running %s", fuzzer)
+        with open(status_path, "w", encoding="utf-8") as status:
+            status.write(
+                f"FAIL\n{stopwatch.start_time_str}\n{stopwatch.duration_seconds}\n"
+            )
+    except subprocess.TimeoutExpired:
+        logging.info("Successful running %s", fuzzer)
+        with open(status_path, "w", encoding="utf-8") as status:
+            status.write(
+                f"OK\n{stopwatch.start_time_str}\n{stopwatch.duration_seconds}\n"
+            )
+    except Exception as e:
+        logging.info("Unexpected exception running %s: %s", fuzzer, e)
+        with open(status_path, "w", encoding="utf-8") as status:
+            status.write(
+                f"ERROR\n{stopwatch.start_time_str}\n{stopwatch.duration_seconds}\n"
+            )
+    else:
+        logging.info("Error running %s", fuzzer)
+        with open(status_path, "w", encoding="utf-8") as status:
+            status.write(
+                f"ERROR\n{stopwatch.start_time_str}\n{stopwatch.duration_seconds}\n"
+            )
 
 
 def main():
@@ -73,10 +149,14 @@ def main():
 
     subprocess.check_call("ls -al", shell=True)
 
+    timeout = 30 if TIMEOUT == 0 else TIMEOUT
+
     with Path() as current:
         for fuzzer in current.iterdir():
             if (current / fuzzer).is_file() and os.access(current / fuzzer, os.X_OK):
-                run_fuzzer(fuzzer)
+                run_fuzzer(fuzzer.name, timeout)
+
+    subprocess.check_call(f"ls -al {OUTPUT}", shell=True)
 
 
 if __name__ == "__main__":

tests/integration/test_named_collections/test.py

@@ -794,3 +794,17 @@ def test_keeper_storage_remove_on_cluster(cluster, ignore, expected_raise):
         node.query(
             f"DROP NAMED COLLECTION test_nc ON CLUSTER `replicated_nc_nodes_cluster`"
         )
+
+
+@pytest.mark.parametrize(
+    "instance_name",
+    [("node"), ("node_with_keeper")],
+)
+def test_name_escaping(cluster, instance_name):
+    node = cluster.instances[instance_name]
+
+    node.query("DROP NAMED COLLECTION IF EXISTS `test_!strange/symbols!`;")
+    node.query("CREATE NAMED COLLECTION `test_!strange/symbols!` AS key1=1, key2=2")
+    node.restart_clickhouse()
+
+    node.query("DROP NAMED COLLECTION `test_!strange/symbols!`")

utils/CMakeLists.txt

@@ -23,3 +23,7 @@ if (ENABLE_UTILS)
     add_subdirectory (keeper-data-dumper)
     add_subdirectory (memcpy-bench)
 endif ()
+
+if (ENABLE_FUZZING AND ENABLE_FUZZER_TEST)
+    add_subdirectory (libfuzzer-test)
+endif ()

utils/libfuzzer-test/CMakeLists.txt

@@ -0,0 +1 @@
+add_subdirectory (test_basic_fuzzer)

utils/libfuzzer-test/README.md

@@ -0,0 +1 @@
+This folder contains various stuff intended to test libfuzzer functionality.

utils/libfuzzer-test/test_basic_fuzzer/CMakeLists.txt

@@ -0,0 +1 @@
+add_executable (test_basic_fuzzer main.cpp)

utils/libfuzzer-test/test_basic_fuzzer/main.cpp

@@ -0,0 +1,11 @@
+#include <stdint.h>
+#include <stddef.h>
+
+extern "C" int LLVMFuzzerTestOneInput(const uint8_t *data, size_t size)
+{
+    if (size > 0 && data[0] == 'H')
+        if (size > 1 && data[1] == 'I')
+            if (size > 2 && data[2] == '!')
+                __builtin_trap();
+    return 0;
+}

utils/list-versions/version_date.tsv

@@ -1,5 +1,7 @@
+v24.10.1.2812-stable	2024-11-01
 v24.9.2.42-stable	2024-10-03
 v24.9.1.3278-stable	2024-09-26
+v24.8.6.70-lts	2024-11-04
 v24.8.5.115-lts	2024-10-08
 v24.8.4.13-lts	2024-09-06
 v24.8.3.59-lts	2024-09-03