Merge branch 'wenxuan/p1/3_dump_checkpoint' of github.com:breezewish/…

…tiflash into wenxuan/p1/3_dump_checkpoint

CMakeLists.txt

@@ -159,7 +159,7 @@ endif ()
 
 include (cmake/cpu_features.cmake)
 
-set (CMAKE_CXX_STANDARD 17)
+set (CMAKE_CXX_STANDARD 20)
 set (CMAKE_CXX_EXTENSIONS 1) # https://cmake.org/cmake/help/latest/prop_tgt/CXX_EXTENSIONS.html#prop_tgt:CXX_EXTENSIONS
 set (CMAKE_CXX_STANDARD_REQUIRED ON)
 

README.md

@@ -41,6 +41,7 @@ The following packages are required:
 - Rust
 - Python 3.0+
 - Ninja-Build or GNU Make
+- Ccache (not necessary but highly recommended to reduce rebuild time)
 
 Detailed steps for each platform are listed below.
 
@@ -55,13 +56,13 @@ curl https://sh.rustup.rs -sSf | sh -s -- --default-toolchain none
 source $HOME/.cargo/env
 
 # Install LLVM, see https://apt.llvm.org for details
-# Clang will be available as /usr/bin/clang++-14
+# Clang will be available as /usr/bin/clang++-15
 wget https://apt.llvm.org/llvm.sh
 chmod +x llvm.sh
-sudo ./llvm.sh 14 all
+sudo ./llvm.sh 15 all
 
 # Install other dependencies
-sudo apt install -y cmake ninja-build zlib1g-dev libcurl4-openssl-dev
+sudo apt install -y cmake ninja-build zlib1g-dev libcurl4-openssl-dev ccache
 ```
 
 **Note for Ubuntu 18.04 and Ubuntu 20.04:**
@@ -97,7 +98,7 @@ curl https://sh.rustup.rs -sSf | sh -s -- --default-toolchain none
 source $HOME/.cargo/env
 
 # Install compilers and dependencies
-sudo pacman -S clang lld libc++ libc++abi compiler-rt openmp lcov cmake ninja curl openssl zlib llvm
+sudo pacman -S clang lld libc++ libc++abi compiler-rt openmp lcov cmake ninja curl openssl zlib llvm ccache
 ```
 
 </details>
@@ -121,7 +122,7 @@ source $HOME/.cargo/env
 xcode-select --install
 
 # Install other dependencies
-brew install ninja cmake openssl@1.1
+brew install ninja cmake openssl@1.1 ccache
 ```
 
 </details>

dbms/CMakeLists.txt

@@ -225,6 +225,7 @@ endif()
 # We mark it public to make the headers available.
 if (Poco_NetSSL_FOUND)
     target_link_libraries (tiflash_common_io PUBLIC ${Poco_NetSSL_LIBRARY})
+    target_no_warning (tiflash_common_io deprecated-enum-enum-conversion)
 endif()
 
 target_link_libraries (dbms ${Poco_Foundation_LIBRARY})

dbms/src/Columns/ColumnsCommon.cpp

@@ -16,6 +16,8 @@
 #include <Columns/IColumn.h>
 #include <common/memcpy.h>
 
+#include <bit>
+
 #ifdef TIFLASH_ENABLE_AVX_SUPPORT
 ASSERT_USE_AVX2_COMPILE_FLAG
 #endif
@@ -106,7 +108,7 @@ static inline size_t CountBytesInFilterWithNull(const IColumn::Filter & filt, co
 #if defined(__SSE2__) || defined(__AVX2__)
     for (; size >= 64;)
     {
-        count += __builtin_popcountll(ToBits64(p1) & ~ToBits64(p2));
+        count += std::popcount(ToBits64(p1) & ~ToBits64(p2));
         p1 += 64, p2 += 64;
         size -= 64;
     }
@@ -278,7 +280,7 @@ void filterArraysImplGeneric(
         {
             while (mask)
             {
-                size_t index = __builtin_ctz(mask);
+                size_t index = std::countr_zero(mask);
                 copy_array(offsets_pos + index);
                 mask = mask & (mask - 1);
             }

dbms/src/Common/Exception.h

@@ -50,7 +50,7 @@ class Exception : public Poco::Exception
     // Format message with fmt::format, like the logging functions.
     template <typename... Args>
     Exception(int code, const std::string & fmt, Args &&... args)
-        : Exception(FmtBuffer().fmtAppend(fmt, std::forward<Args>(args)...).toString(), code)
+        : Exception(FmtBuffer().fmtAppend(fmt::runtime(fmt), std::forward<Args>(args)...).toString(), code)
     {}
 
     Exception(const std::string & msg, const std::string & arg, int code = 0)
@@ -184,7 +184,7 @@ inline std::string generateFormattedMessage(const char * condition)
 template <typename... Args>
 inline std::string generateFormattedMessage(const char * condition, const char * fmt_str, Args &&... args)
 {
-    return FmtBuffer().fmtAppend("Assert {} fail! ", condition).fmtAppend(fmt_str, std::forward<Args>(args)...).toString();
+    return FmtBuffer().fmtAppend("Assert {} fail! ", condition).fmtAppend(fmt::runtime(fmt_str), std::forward<Args>(args)...).toString();
 }
 
 template <typename... Args>

dbms/src/Common/FailPoint.cpp

@@ -120,6 +120,7 @@ namespace DB
 
 #define APPLY_FOR_RANDOM_FAILPOINTS(M)                  \
     M(random_tunnel_wait_timeout_failpoint)             \
+    M(random_tunnel_write_failpoint)                    \
     M(random_tunnel_init_rpc_failure_failpoint)         \
     M(random_receiver_local_msg_push_failure_failpoint) \
     M(random_receiver_sync_msg_push_failure_failpoint)  \
@@ -132,7 +133,13 @@ namespace DB
     M(random_sharedquery_failpoint)                     \
     M(random_interpreter_failpoint)                     \
     M(random_task_manager_find_task_failure_failpoint)  \
-    M(random_min_tso_scheduler_failpoint)
+    M(random_min_tso_scheduler_failpoint)               \
+    M(random_pipeline_model_task_run_failpoint)         \
+    M(random_pipeline_model_task_construct_failpoint)   \
+    M(random_pipeline_model_event_schedule_failpoint)   \
+    M(random_pipeline_model_event_finish_failpoint)     \
+    M(random_pipeline_model_operator_run_failpoint)     \
+    M(random_pipeline_model_cancel_failpoint)
 
 namespace FailPoints
 {

dbms/src/Common/HyperLogLogCounter.h

@@ -24,6 +24,7 @@
 #include <IO/WriteHelpers.h>
 #include <common/types.h>
 
+#include <bit>
 #include <cmath>
 #include <cstring>
 
@@ -69,7 +70,7 @@ struct LogLUT
 private:
     static constexpr size_t M = 1 << ((static_cast<unsigned int>(K) <= 12) ? K : 12);
 
-    double log_table[M + 1];
+    double log_table[M + 1]{};
 };
 
 template <UInt8 K>
@@ -233,7 +234,7 @@ struct TrailingZerosCounter<UInt32>
 {
     static int apply(UInt32 val)
     {
-        return __builtin_ctz(val);
+        return std::countr_zero(val);
     }
 };
 
@@ -242,7 +243,7 @@ struct TrailingZerosCounter<UInt64>
 {
     static int apply(UInt64 val)
     {
-        return __builtin_ctzll(val);
+        return std::countr_zero(val);
     }
 };
 
@@ -439,7 +440,7 @@ class __attribute__((packed)) HyperLogLogCounter : private Hash
     void update(HashValueType bucket, UInt8 rank)
     {
         typename RankStore::Locus content = rank_store[bucket];
-        UInt8 cur_rank = static_cast<UInt8>(content);
+        auto cur_rank = static_cast<UInt8>(content);
 
         if (rank > cur_rank)
         {

dbms/src/Common/MyDuration.cpp

@@ -65,6 +65,6 @@ String MyDuration::toString() const
     }
     auto fmt_str = fmt::format("{}{}{}", "{}{:02}:{:02}:{:02}.{:.", fsp, "}");
     auto frac_str = fmt::format("{:06}", microsecond);
-    return FmtBuffer().fmtAppend(fmt_str, sign > 0 ? "" : "-", hour, minute, second, frac_str).toString();
+    return FmtBuffer().fmtAppend(fmt::runtime(fmt_str), sign > 0 ? "" : "-", hour, minute, second, frac_str).toString();
 }
 } // namespace DB

dbms/src/Common/ProfileEvents.cpp

@@ -121,7 +121,9 @@
     M(S3CompleteMultipartUpload)               \
     M(S3PutObject)                             \
     M(S3GetObject)                             \
-    M(S3HeadObject)
+    M(S3HeadObject)                            \
+    M(S3ListObjects)                           \
+    M(S3DeleteObject)
 
 namespace ProfileEvents
 {

dbms/src/Common/StringSearcher.h

@@ -21,6 +21,7 @@
 #include <stdint.h>
 #include <string.h>
 
+#include <bit>
 #include <ext/range.h>
 
 #if __SSE2__
@@ -248,7 +249,7 @@ class StringSearcher<false, false> : private StringSearcherBase
                 const auto v_against_u = _mm_cmpeq_epi8(v_haystack, patu);
                 const auto v_against_l_or_u = _mm_or_si128(v_against_l, v_against_u);
 
-                const auto mask = _mm_movemask_epi8(v_against_l_or_u);
+                const UInt32 mask = _mm_movemask_epi8(v_against_l_or_u);
 
                 if (mask == 0)
                 {
@@ -257,7 +258,7 @@ class StringSearcher<false, false> : private StringSearcherBase
                     continue;
                 }
 
-                const auto offset = __builtin_ctz(mask);
+                const auto offset = std::countr_zero(mask);
                 haystack += offset;
 
                 if (haystack < haystack_end && haystack + n <= haystack_end && pageSafe(haystack))
@@ -444,15 +445,15 @@ class StringSearcher<false, true> : private StringSearcherBase
                 const auto v_against_u = _mm_cmpeq_epi8(v_haystack, patu);
                 const auto v_against_l_or_u = _mm_or_si128(v_against_l, v_against_u);
 
-                const auto mask = _mm_movemask_epi8(v_against_l_or_u);
+                const UInt32 mask = _mm_movemask_epi8(v_against_l_or_u);
 
                 if (mask == 0)
                 {
                     haystack += n;
                     continue;
                 }
 
-                const auto offset = __builtin_ctz(mask);
+                const auto offset = std::countr_zero(mask);
                 haystack += offset;
 
                 if (haystack < haystack_end && haystack + n <= haystack_end && pageSafe(haystack))
@@ -622,7 +623,7 @@ class StringSearcher<true, ASCII> : private StringSearcherBase
                 const auto v_haystack = _mm_loadu_si128(reinterpret_cast<const __m128i *>(haystack));
                 const auto v_against_pattern = _mm_cmpeq_epi8(v_haystack, pattern);
 
-                const auto mask = _mm_movemask_epi8(v_against_pattern);
+                const UInt32 mask = _mm_movemask_epi8(v_against_pattern);
 
                 /// first character not present in 16 octets starting at `haystack`
                 if (mask == 0)
@@ -631,7 +632,7 @@ class StringSearcher<true, ASCII> : private StringSearcherBase
                     continue;
                 }
 
-                const auto offset = __builtin_ctz(mask);
+                const auto offset = std::countr_zero(mask);
                 haystack += offset;
 
                 if (haystack < haystack_end && haystack + n <= haystack_end && pageSafe(haystack))

dbms/src/Common/TiFlashException.h

@@ -253,7 +253,7 @@ class TiFlashException : public Exception
 
     template <typename... Args>
     TiFlashException(const TiFlashError & _error, const std::string & fmt, Args &&... args)
-        : Exception(FmtBuffer().fmtAppend(fmt, std::forward<Args>(args)...).toString())
+        : Exception(FmtBuffer().fmtAppend(fmt::runtime(fmt), std::forward<Args>(args)...).toString())
         , error(_error)
     {}
 

dbms/src/Common/TiFlashMetrics.h

@@ -286,11 +286,21 @@ namespace DB
         F(type_mpp, {{"type", "mpp"}, ComputeLabelHolder::instance().getClusterIdLabel(), ComputeLabelHolder::instance().getProcessIdLabel()}),     \
         F(type_cop, {{"type", "cop"}, ComputeLabelHolder::instance().getClusterIdLabel(), ComputeLabelHolder::instance().getProcessIdLabel()}),     \
         F(type_batch, {{"type", "batch"}, ComputeLabelHolder::instance().getClusterIdLabel(), ComputeLabelHolder::instance().getProcessIdLabel()})) \
-    M(tiflash_shared_block_schemas, "statistics about shared block schemas of ColumnFiles", Gauge,                                                                 \
+    M(tiflash_shared_block_schemas, "statistics about shared block schemas of ColumnFiles", Gauge,                                                  \
         F(type_current_size, {{"type", "current_size"}}),                                                                                           \
-        F(type_still_used_when_evict, {{"type", "still_used_when_evict"}}),                                                                    \
-        F(type_miss_count, {{"type", "miss_count"}}),                                                                                          \
-        F(type_hit_count, {{"type", "hit_count"}}))
+        F(type_still_used_when_evict, {{"type", "still_used_when_evict"}}),                                                                         \
+        F(type_miss_count, {{"type", "miss_count"}}),                                                                                               \
+        F(type_hit_count, {{"type", "hit_count"}}))                                                                                                 \
+    M(tiflash_storage_s3_request_seconds, "S3 request duration in seconds", Histogram,                                                              \
+        F(type_put_object, {{"type", "put_object"}}, ExpBuckets{0.001, 2, 20}),                                                                     \
+        F(type_get_object, {{"type", "get_object"}}, ExpBuckets{0.001, 2, 20}),                                                                     \
+        F(type_create_multi_part_upload, {{"type", "create_multi_part_upload"}}, ExpBuckets{0.001, 2, 20}),                                         \
+        F(type_upload_part, {{"type", "upload_part"}}, ExpBuckets{0.001, 2, 20}),                                                                   \
+        F(type_complete_multi_part_upload, {{"type", "complete_multi_part_upload"}}, ExpBuckets{0.001, 2, 20}),                                     \
+        F(type_list_objects, {{"type", "list_objects"}}, ExpBuckets{0.001, 2, 20}),                                                                 \
+        F(type_delete_object, {{"type", "delete_object"}}, ExpBuckets{0.001, 2, 20}),                                                               \
+        F(type_head_object, {{"type", "head_object"}}, ExpBuckets{0.001, 2, 20}))
+
 // clang-format on
 
 /// Buckets with boundaries [start * base^0, start * base^1, ..., start * base^(size-1)]

dbms/src/DataStreams/DistinctBlockInputStream.cpp

@@ -19,7 +19,7 @@ namespace DB
 
 namespace ErrorCodes
 {
-    extern const int SET_SIZE_LIMIT_EXCEEDED;
+extern const int SET_SIZE_LIMIT_EXCEEDED;
 }
 
 DistinctBlockInputStream::DistinctBlockInputStream(const BlockInputStreamPtr & input, const SizeLimits & set_size_limits, size_t limit_hint_, const Names & columns)
@@ -34,7 +34,7 @@ Block DistinctBlockInputStream::readImpl()
 {
     /// Execute until end of stream or until
     /// a block with some new records will be gotten.
-    while (1)
+    while (true)
     {
         if (no_more_rows)
             return Block();
@@ -66,14 +66,15 @@ Block DistinctBlockInputStream::readImpl()
 
         switch (data.type)
         {
-            case SetVariants::Type::EMPTY:
-                break;
-        #define M(NAME) \
-            case SetVariants::Type::NAME: \
-                buildFilter(*data.NAME, column_ptrs, filter, rows, data); \
-                break;
+            using enum SetVariants::Type;
+        case EMPTY:
+            break;
+#define M(NAME)                                                   \
+    case NAME:                                                    \
+        buildFilter(*data.NAME, column_ptrs, filter, rows, data); \
+        break;
             APPLY_FOR_SET_VARIANTS(M)
-        #undef M
+#undef M
         }
 
         /// Just go to the next block if there isn't any new record in the current one.
@@ -123,7 +124,7 @@ ColumnRawPtrs DistinctBlockInputStream::getKeyColumns(const Block & block) const
 
     for (size_t i = 0; i < columns; ++i)
     {
-        auto & column = columns_names.empty()
+        const auto & column = columns_names.empty()
             ? block.safeGetByPosition(i).column
             : block.getByName(columns_names[i]).column;
 
@@ -135,4 +136,4 @@ ColumnRawPtrs DistinctBlockInputStream::getKeyColumns(const Block & block) const
     return column_ptrs;
 }
 
-}
+} // namespace DB