[Improvement](shuffle) add Crc32CHashPartitioner (#59052)

add Crc32CHashPartitioner
<img width="596" height="4284" alt="图片"
src="https://github.com/user-attachments/assets/5773ea04-b01a-4c8c-ba5a-0c725cb11f11"
/>
This pull request refactors the codebase to standardize the usage of the
CRC32C checksum library by replacing the custom `util/crc32c.h` header
and its functions with the upstream `crc32c` library
(`<crc32c/crc32c.h>`) and its API. It also updates function calls to use
the correct data types expected by the new library and ensures
consistent checksum calculation across multiple modules related to file
I/O, compression, and storage.

**Migration to Upstream CRC32C Library**

* Replaced all includes of `"util/crc32c.h"` with `<crc32c/crc32c.h>`
and removed the custom header from all relevant files.
[[1]](diffhunk://#diff-0572424f9b6fe1561e15b070c1155b1b8f9272499029d425ff5a8d0e0aa8f40fL24)
[[2]](diffhunk://#diff-a4327d67c48e4a4115a1ac9bc0a82a646bbfcb141d80f5f428142f55027e16a1R20-L21)
[[3]](diffhunk://#diff-f46297d8957a9929f575febc300a004c144e106ea6893f1b95508ab006503407R18-L23)
[[4]](diffhunk://#diff-3ef6a4f806adc33273c229fbdb827c072152651d5930b19affde4c1f8984c51cR20-L23)
[[5]](diffhunk://#diff-52cdc310f4ed34081299dff53c543455745a834dbc5c50a2c21b765c0c90c3f8R18)
[[6]](diffhunk://#diff-03a87568e2651d1524985a56a278f2e2932667c1e92efc60d0c5a750f0ad316bR20-R21)
[[7]](diffhunk://#diff-23fa0193d626ba712c4186c66bcd1809c7e55bfc04ea10f5a91c691ed3e04727R21)
[[8]](diffhunk://#diff-4dc7440cc992e7f9bdd8ec9c5bfc5a6194f9d78fc5ff359c4781d992df4e610bR20)
[[9]](diffhunk://#diff-5eb6e846447db952b75ba0fd9bc1614702c428689c93e089a952ea414c23b7fdR20)
[[10]](diffhunk://#diff-c33a6f975ebaa66163e68ba51a4d9ce0cbfd6b5d063edce503130d7bae502c53R20)
[[11]](diffhunk://#diff-8061bb86d18c96049b63aa2caf4851933bff6b16cefa5460b1ee736d6f0ac883R27-R28)
[[12]](diffhunk://#diff-9018eae3f9bef2cf64079552ce4d9c3fd3535a31b86a4ff496d29853c4968cb0R20)
* Updated all function calls from `crc32c::Value(...)` to
`crc32c::Crc32c(...)` for computing CRC32C checksums.
[[1]](diffhunk://#diff-0572424f9b6fe1561e15b070c1155b1b8f9272499029d425ff5a8d0e0aa8f40fL120-R119)
[[2]](diffhunk://#diff-a4327d67c48e4a4115a1ac9bc0a82a646bbfcb141d80f5f428142f55027e16a1L89-R90)
[[3]](diffhunk://#diff-3ef6a4f806adc33273c229fbdb827c072152651d5930b19affde4c1f8984c51cL189-R190)
[[4]](diffhunk://#diff-3ef6a4f806adc33273c229fbdb827c072152651d5930b19affde4c1f8984c51cL420-R421)
[[5]](diffhunk://#diff-52cdc310f4ed34081299dff53c543455745a834dbc5c50a2c21b765c0c90c3f8L180-R180)
[[6]](diffhunk://#diff-ea6232df0f48fea9e5403472da0bc4206acfd69b676c1b5fbc2d2df13df24624L149-R150)
[[7]](diffhunk://#diff-5eb6e846447db952b75ba0fd9bc1614702c428689c93e089a952ea414c23b7fdL178-R181)
[[8]](diffhunk://#diff-c33a6f975ebaa66163e68ba51a4d9ce0cbfd6b5d063edce503130d7bae502c53L472-R472)
[[9]](diffhunk://#diff-8061bb86d18c96049b63aa2caf4851933bff6b16cefa5460b1ee736d6f0ac883L1158-R1159)
* Updated all function calls from `crc32c::Extend(...)` to use the new
function signature, casting data pointers to `const uint8_t*` as
required by the upstream library.
[[1]](diffhunk://#diff-f46297d8957a9929f575febc300a004c144e106ea6893f1b95508ab006503407L320-R321)
[[2]](diffhunk://#diff-f46297d8957a9929f575febc300a004c144e106ea6893f1b95508ab006503407L369-R370)
[[3]](diffhunk://#diff-ea6232df0f48fea9e5403472da0bc4206acfd69b676c1b5fbc2d2df13df24624L103-R104)
[[4]](diffhunk://#diff-23fa0193d626ba712c4186c66bcd1809c7e55bfc04ea10f5a91c691ed3e04727L2037-R2037)
[[5]](diffhunk://#diff-4dc7440cc992e7f9bdd8ec9c5bfc5a6194f9d78fc5ff359c4781d992df4e610bL734-R734)

**Checksum Calculation Logic**

* Modified checksum calculation for multi-slice data by iteratively
using `crc32c::Extend` over each slice, ensuring correct cumulative
checksum computation.
* Updated checksum verification logic to use the new API and data types,
improving reliability and consistency across modules.
[[1]](diffhunk://#diff-0572424f9b6fe1561e15b070c1155b1b8f9272499029d425ff5a8d0e0aa8f40fL120-R119)
[[2]](diffhunk://#diff-a4327d67c48e4a4115a1ac9bc0a82a646bbfcb141d80f5f428142f55027e16a1L89-R90)
[[3]](diffhunk://#diff-3ef6a4f806adc33273c229fbdb827c072152651d5930b19affde4c1f8984c51cL189-R190)
[[4]](diffhunk://#diff-3ef6a4f806adc33273c229fbdb827c072152651d5930b19affde4c1f8984c51cL420-R421)
[[5]](diffhunk://#diff-52cdc310f4ed34081299dff53c543455745a834dbc5c50a2c21b765c0c90c3f8L180-R180)
[[6]](diffhunk://#diff-ea6232df0f48fea9e5403472da0bc4206acfd69b676c1b5fbc2d2df13df24624L149-R150)
[[7]](diffhunk://#diff-5eb6e846447db952b75ba0fd9bc1614702c428689c93e089a952ea414c23b7fdL178-R181)
[[8]](diffhunk://#diff-c33a6f975ebaa66163e68ba51a4d9ce0cbfd6b5d063edce503130d7bae502c53L472-R472)
[[9]](diffhunk://#diff-8061bb86d18c96049b63aa2caf4851933bff6b16cefa5460b1ee736d6f0ac883L1158-R1159)

**Code Clean-up and Consistency**

* Removed all redundant or obsolete includes of the custom `crc32c.h`
header.
[[1]](diffhunk://#diff-a4327d67c48e4a4115a1ac9bc0a82a646bbfcb141d80f5f428142f55027e16a1R20-L21)
[[2]](diffhunk://#diff-f46297d8957a9929f575febc300a004c144e106ea6893f1b95508ab006503407R18-L23)
[[3]](diffhunk://#diff-3ef6a4f806adc33273c229fbdb827c072152651d5930b19affde4c1f8984c51cR20-L23)
[[4]](diffhunk://#diff-52cdc310f4ed34081299dff53c543455745a834dbc5c50a2c21b765c0c90c3f8L30)
[[5]](diffhunk://#diff-03a87568e2651d1524985a56a278f2e2932667c1e92efc60d0c5a750f0ad316bL30)
[[6]](diffhunk://#diff-23fa0193d626ba712c4186c66bcd1809c7e55bfc04ea10f5a91c691ed3e04727L51)
[[7]](diffhunk://#diff-4dc7440cc992e7f9bdd8ec9c5bfc5a6194f9d78fc5ff359c4781d992df4e610bL49)
[[8]](diffhunk://#diff-5eb6e846447db952b75ba0fd9bc1614702c428689c93e089a952ea414c23b7fdL44)
[[9]](diffhunk://#diff-c33a6f975ebaa66163e68ba51a4d9ce0cbfd6b5d063edce503130d7bae502c53L69)
[[10]](diffhunk://#diff-8061bb86d18c96049b63aa2caf4851933bff6b16cefa5460b1ee736d6f0ac883L65)
* Ensured all modules that require CRC32C now directly depend on the
upstream library, reducing maintenance overhead and potential for bugs.
(all above references)

These changes collectively improve code maintainability, reliability,
and alignment with upstream best practices for CRC32C checksum
operations.

Author: BiteTheDDDDt

be/src/cloud/delete_bitmap_file_reader.cpp

@@ -21,7 +21,6 @@
 #include "common/status.h"
 #include "io/fs/file_reader.h"
 #include "util/coding.h"
-#include "util/crc32c.h"
 
 namespace doris {
 #include "common/compile_check_begin.h"
@@ -117,7 +116,7 @@ Status DeleteBitmapFileReader::read(DeleteBitmapPB& delete_bitmap) {
             offset, {checksum_len_buf, DeleteBitmapFileWriter::CHECKSUM_SIZE}, &bytes_read));
     offset += DeleteBitmapFileWriter::CHECKSUM_SIZE;
     uint32_t checksum = decode_fixed32_le(checksum_len_buf);
-    uint32_t computed_checksum = crc32c::Value(delete_bitmap_buf.data(), delete_bitmap_len);
+    uint32_t computed_checksum = crc32c::Crc32c(delete_bitmap_buf.data(), delete_bitmap_len);
     if (computed_checksum != checksum) {
         return Status::InternalError("delete bitmap checksum failed from file=" + _path +
                                      ", computed checksum=" + std::to_string(computed_checksum) +

be/src/cloud/delete_bitmap_file_writer.cpp

@@ -17,8 +17,9 @@
 
 #include "cloud/delete_bitmap_file_writer.h"
 
+#include <crc32c/crc32c.h>
+
 #include "io/fs/file_writer.h"
-#include "util/crc32c.h"
 
 namespace doris {
 #include "common/compile_check_begin.h"
@@ -86,7 +87,7 @@ Status DeleteBitmapFileWriter::write(const DeleteBitmapPB& delete_bitmap) {
 
     // 3. write checksum
     uint8_t checksum_buf[CHECKSUM_SIZE];
-    uint32_t checksum = crc32c::Value(content.data(), delete_bitmap_len);
+    uint32_t checksum = crc32c::Crc32c(content.data(), delete_bitmap_len);
     encode_fixed32_le(checksum_buf, checksum);
     RETURN_IF_ERROR(_file_writer->append({checksum_buf, CHECKSUM_SIZE}));
     return Status::OK();

be/src/exec/lzo_decompressor.cpp

@@ -15,12 +15,13 @@
 // specific language governing permissions and limitations
 // under the License.
 
+#include <crc32c/crc32c.h>
+
 #include "common/cast_set.h"
 #include "common/logging.h"
 #include "exec/decompressor.h"
 #include "olap/utils.h"
 #include "orc/Exceptions.hh"
-#include "util/crc32c.h"
 
 namespace orc {
 /**
@@ -317,7 +318,7 @@ Status LzopDecompressor::parse_header_info(uint8_t* input, size_t input_len,
     uint32_t computed_checksum;
     if (_header_info.header_checksum_type == CHECK_CRC32) {
         computed_checksum = CRC32_INIT_VALUE;
-        computed_checksum = crc32c::Extend(computed_checksum, (const char*)header, cur - header);
+        computed_checksum = crc32c::Extend(computed_checksum, (const uint8_t*)header, cur - header);
     } else {
         computed_checksum = ADLER32_INIT_VALUE;
         computed_checksum = olap_adler32(computed_checksum, (const char*)header, cur - header);
@@ -366,7 +367,7 @@ Status LzopDecompressor::checksum(LzoChecksum type, const std::string& source, u
     case CHECK_NONE:
         return Status::OK();
     case CHECK_CRC32:
-        computed_checksum = crc32c::Extend(CRC32_INIT_VALUE, (const char*)ptr, len);
+        computed_checksum = crc32c::Extend(CRC32_INIT_VALUE, (const uint8_t*)ptr, len);
         break;
     case CHECK_ADLER:
         computed_checksum = olap_adler32(ADLER32_INIT_VALUE, (const char*)ptr, len);

be/src/exprs/block_bloom_filter.hpp

@@ -20,6 +20,8 @@
 
 #pragma once
 
+#include <crc32c/crc32c.h>
+
 #include "vec/common/string_ref.h"
 #ifdef __AVX2__
 #include <immintrin.h>
@@ -28,7 +30,6 @@
 #endif
 
 #include "common/status.h"
-#include "util/hash_util.hpp"
 #include "util/slice.h"
 
 namespace butil {
@@ -76,7 +77,7 @@ class BlockBloomFilter {
     // Same as above with convenience of hashing the key.
     void insert(const StringRef& key) noexcept {
         if (key.data) {
-            insert(HashUtil::crc32c_hash(key.data, uint32_t(key.size), _hash_seed));
+            insert(crc32c::Extend(_hash_seed, (const uint8_t*)key.data, uint32_t(key.size)));
         }
     }
 
@@ -105,7 +106,7 @@ class BlockBloomFilter {
     // Same as above with convenience of hashing the key.
     bool find(const StringRef& key) const noexcept {
         if (key.data) {
-            return find(HashUtil::crc32c_hash(key.data, uint32_t(key.size), _hash_seed));
+            return find(crc32c::Extend(_hash_seed, (const uint8_t*)key.data, uint32_t(key.size)));
         }
         return false;
     }

be/src/io/cache/cache_lru_dumper.cpp

@@ -17,10 +17,11 @@
 
 #include "io/cache/cache_lru_dumper.h"
 
+#include <crc32c/crc32c.h>
+
 #include "io/cache/block_file_cache.h"
 #include "io/cache/lru_queue_recorder.h"
 #include "util/coding.h"
-#include "util/crc32c.h"
 #include "vec/common/endian.h"
 
 namespace doris::io {
@@ -186,7 +187,7 @@ Status CacheLRUDumper::flush_current_group(std::ofstream& out, std::string& file
     ::doris::io::cache::EntryGroupOffsetSizePb* group_info = _dump_meta.add_group_offset_size();
     group_info->set_offset(group_start);
     group_info->set_size(serialized.size());
-    uint32_t checksum = crc32c::Value(serialized.data(), serialized.size());
+    uint32_t checksum = crc32c::Crc32c(serialized.data(), serialized.size());
     group_info->set_checksum(checksum);
 
     // Reset for next group
@@ -417,7 +418,7 @@ Status CacheLRUDumper::parse_one_lru_entry(std::ifstream& in, std::string& filen
         std::string group_serialized(group_info.size(), '\0');
         in.read(&group_serialized[0], group_serialized.size());
         RETURN_IF_ERROR(check_ifstream_status(in, filename));
-        uint32_t checksum = crc32c::Value(group_serialized.data(), group_serialized.size());
+        uint32_t checksum = crc32c::Crc32c(group_serialized.data(), group_serialized.size());
         if (checksum != group_info.checksum()) {
             std::string warn_msg =
                     fmt::format("restore lru failed as checksum not match, file={}", filename);

be/src/io/cache/file_cache_lru_tool.cpp

@@ -15,6 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
+#include <crc32c/crc32c.h>
 #include <gflags/gflags.h>
 
 #include <fstream>
@@ -27,7 +28,6 @@
 #include "io/cache/file_cache_common.h"
 #include "io/cache/lru_queue_recorder.h"
 #include "util/coding.h"
-#include "util/crc32c.h"
 
 using namespace doris;
 
@@ -177,7 +177,7 @@ Status parse_one_lru_entry(std::ifstream& in, std::string& filename, io::UInt128
         std::string group_serialized(group_info.size(), '\0');
         in.read(&group_serialized[0], group_serialized.size());
         RETURN_IF_ERROR(check_ifstream_status(in, filename));
-        uint32_t checksum = crc32c::Value(group_serialized.data(), group_serialized.size());
+        uint32_t checksum = crc32c::Crc32c(group_serialized.data(), group_serialized.size());
         if (checksum != group_info.checksum()) {
             std::string warn_msg =
                     fmt::format("restore lru failed as checksum not match, file={}", filename);

be/src/io/fs/s3_file_bufferpool.cpp

@@ -18,6 +18,7 @@
 #include "s3_file_bufferpool.h"
 
 #include <bvar/bvar.h>
+#include <crc32c/crc32c.h>
 
 #include <chrono>
 #include <memory>
@@ -100,7 +101,7 @@ Status UploadFileBuffer::append_data(const Slice& data) {
                                       data.get_size());
     std::memcpy((void*)(_inner_data->data().get_data() + _size), data.get_data(), data.get_size());
     _size += data.get_size();
-    _crc_value = crc32c::Extend(_crc_value, data.get_data(), data.get_size());
+    _crc_value = crc32c::Extend(_crc_value, (const uint8_t*)data.get_data(), data.get_size());
     return Status::OK();
 }
 
@@ -146,7 +147,7 @@ std::string_view FileBuffer::get_string_view_data() const {
 
 void UploadFileBuffer::on_upload() {
     _stream_ptr = std::make_shared<StringViewStream>(_inner_data->data().get_data(), _size);
-    if (_crc_value != crc32c::Value(_inner_data->data().get_data(), _size)) {
+    if (_crc_value != crc32c::Crc32c(_inner_data->data().get_data(), _size)) {
         DCHECK(false);
         set_status(Status::IOError("Buffer checksum not match"));
         return;

be/src/io/fs/s3_file_bufferpool.h

@@ -17,6 +17,8 @@
 
 #pragma once
 
+#include <crc32c/crc32c.h>
+
 #include <condition_variable>
 #include <cstdint>
 #include <fstream>
@@ -27,7 +29,6 @@
 
 #include "common/status.h"
 #include "io/cache/file_block.h"
-#include "util/crc32c.h"
 #include "util/slice.h"
 #include "util/threadpool.h"
 

be/src/olap/base_tablet.cpp

@@ -18,6 +18,7 @@
 #include "olap/base_tablet.h"
 
 #include <bthread/mutex.h>
+#include <crc32c/crc32c.h>
 #include <fmt/format.h>
 #include <rapidjson/prettywriter.h>
 
@@ -48,7 +49,6 @@
 #include "olap/txn_manager.h"
 #include "service/point_query_executor.h"
 #include "util/bvar_helper.h"
-#include "util/crc32c.h"
 #include "util/debug_points.h"
 #include "util/doris_metrics.h"
 #include "util/key_util.h"
@@ -2024,7 +2024,7 @@ Status BaseTablet::calc_file_crc(uint32_t* crc_value, int64_t start_version, int
             return st;
         }
         // crc_value is calculated based on the crc_value of each rowset.
-        *crc_value = crc32c::Extend(*crc_value, reinterpret_cast<const char*>(&rs_crc_value),
+        *crc_value = crc32c::Extend(*crc_value, reinterpret_cast<const uint8_t*>(&rs_crc_value),
                                     sizeof(rs_crc_value));
         *file_count += rs_file_count;
     }

be/src/olap/rowset/beta_rowset.cpp

@@ -17,6 +17,7 @@
 
 #include "olap/rowset/beta_rowset.h"
 
+#include <crc32c/crc32c.h>
 #include <ctype.h>
 #include <errno.h>
 #include <fmt/format.h>
@@ -46,7 +47,6 @@
 #include "olap/segment_loader.h"
 #include "olap/tablet_schema.h"
 #include "olap/utils.h"
-#include "util/crc32c.h"
 #include "util/debug_points.h"
 #include "util/doris_metrics.h"
 
@@ -731,7 +731,7 @@ Status BetaRowset::calc_file_crc(uint32_t* crc_value, int64_t* file_count) {
     // 3. calculate the crc_value based on all_file_md5
     DCHECK(file_paths.size() == all_file_md5.size());
     for (auto& i : all_file_md5) {
-        *crc_value = crc32c::Extend(*crc_value, i.data(), i.size());
+        *crc_value = crc32c::Extend(*crc_value, (const uint8_t*)i.data(), i.size());
     }
 
     return Status::OK();