What problem does this PR solve?

Issue Number: close #6620

Problem Summary:

• Optimize exchange operator
• Support mpp-version in mpp tasks

What is changed and how it works?

Base Modules & Utils

• add -march=haswell flag for a few modules
• add avx2_byte_count to optimize size_t countBytesInFilter(const UInt8 * filt, size_t sz).
• optimize a few implementations for ColumnsCommon.cpp, DataTypeString.cpp,
• use std::chrono::steady_clock instead of system_clock in a few modules

Mpp Version Check

TiFlash will check mpp-version in FlashService::EstablishMPPConnection, FlashService::CancelMPPTask, FlashService::EstablishMPPConnection.

if failed to check mpp-version is valid, tiflash will return grpc error grpc::StatusCode::CANCELLED with message like:

ERROR 1105 (HY000): rpc error: code = Canceled desc = Failed to handling mpp dispatch request, reason=`Invalid mpp version -1, TiFlash expects version: min 0, max 1, should upgrade TiDB/planner`

Hash Partition Writer

• add new module HashPartitionWriterV1 with new codec pattern
  • create packet with version MPPDataPacketV1 (1) https://github.com/pingcap/kvproto/blob/7cd28226c2a21e489b077a87e1f5c9ba2c950944/proto/mpp.proto#L76
  • data packet chunk format

            // compression method flag; NONE = 0x02, LZ4 = 0x82, ZSTD = 0x90
            // ... 
            // header meta:
            //     columns count;
            //     total row count (multi parts);
            //     for each column:
            //         column name;
            //         column type;
            // for each part:
            //     row count;
            //     columns data;

• HashPartitionWriterV1 supports compression mode:
  • NONE: no compression
    • if target tunnel model is TunnelSenderMode::LOCAL, must use method NONE.
  • FAST: fast compression/decompression speed, use method LZ4.
  • HIGH_COMPRESSION: high compression (HC) ratio mode, use method ZSTD.
• NewMPPExchangeWriter will generate HashPartitionWriterV1 when exchange type is tipb::ExchangeType::Hash and mpp-version of dag context is not 0.
• ExchangeReceiverBase<RPCContext>::decodeChunks will check version of mpp::MPPDataPacket
  • MPPDataPacketV0: use previous way.
  • MPPDataPacketV1: use new format. check first byte(compression method flag) to determine if data is compressed.

Grafana

Add panel Exchange Bytes/Seconds in Grafana

• hash_none_compression_remote
• hash_none_compression_local
• hash_lz4_compression
• hash_zstd_compression
• hash_original: original data size sent by hash exchange
  • hash_original_all ≈ hash_none_remote + hash_none_local + hash_lz4 * lz4-compress-rate + hash_zstd * zstd-compress-rate
• broadcast_passthrough_original
• broadcast_passthrough_none_compression_local
• broadcast_passthrough_none_compression_remote

Config

[profiles]
[profiles.default]
batch_send_min_limit_compression = -1 # default minimal chunk size of exchanging data among TiFlash when using data compression

If batch_send_min_limit_compression LT 0, HashPartitionWriterV1 will use 8192 * partition_num.

Benchmark

• TPCH-100
• TiFlash x 3
  • Intel(R) Xeon(R) CPU E5-2630 v4 @ 2.20GHz
  • 20C 40T
• TIFLASH REPLICA x 3
• Base branch:
  • TiDB: f1a744a3e6da9bb29781efc5bd504be5cc0ba55c

Notice: in the master branch, #6713 will cause performance regression, try to resolve it and get benchmark below

Check List

Tests

• Unit test
• Integration test
• Manual test (add detailed scripts or steps below)
• No code

Side effects

• Performance regression: Consumes more CPU
• Performance regression: Consumes more Memory
• Breaking backward compatibility

Documentation

• Affects user behaviors
• Contains syntax changes
• Contains variable changes
• Contains experimental features
• Changes MySQL compatibility

Release note

TiFlash supports data compression in MPP Exchange operators