[Improvement](shuffle) add Crc32CHashPartitioner

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"
@@ -2034,7 +2034,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();

be/src/olap/rowset/segment_v2/page_io.cpp

@@ -17,6 +17,7 @@
 
 #include "olap/rowset/segment_v2/page_io.h"
 
+#include <crc32c/crc32c.h>
 #include <gen_cpp/segment_v2.pb.h>
 #include <stdint.h>
 
@@ -41,7 +42,6 @@
 #include "olap/rowset/segment_v2/page_handle.h"
 #include "util/block_compression.h"
 #include "util/coding.h"
-#include "util/crc32c.h"
 #include "util/faststring.h"
 #include "util/runtime_profile.h"
 
@@ -103,7 +103,10 @@ Status PageIO::write_page(io::FileWriter* writer, const std::vector<Slice>& body
 
     // checksum
     uint8_t checksum_buf[sizeof(uint32_t)];
-    uint32_t checksum = crc32c::Value(page);
+    uint32_t checksum = 0;
+    for (const auto& slice : page) {
+        checksum = crc32c::Extend(checksum, (const uint8_t*)slice.data, slice.size);
+    }
     encode_fixed32_le(checksum_buf, checksum);
     page.emplace_back(checksum_buf, sizeof(uint32_t));
 
@@ -175,7 +178,7 @@ Status PageIO::read_and_decompress_page_(const PageReadOptions& opts, PageHandle
 
     if (opts.verify_checksum) {
         uint32_t expect = decode_fixed32_le((uint8_t*)page_slice.data + page_slice.size - 4);
-        uint32_t actual = crc32c::Value(page_slice.data, page_slice.size - 4);
+        uint32_t actual = crc32c::Crc32c(page_slice.data, page_slice.size - 4);
         // here const_cast is used for testing.
         InjectionContext ctx = {&actual, const_cast<PageReadOptions*>(&opts)};
         (void)ctx;

be/src/olap/rowset/segment_v2/segment.cpp

@@ -17,6 +17,7 @@
 
 #include "olap/rowset/segment_v2/segment.h"
 
+#include <crc32c/crc32c.h>
 #include <gen_cpp/Descriptors_types.h>
 #include <gen_cpp/PlanNodes_types.h>
 #include <gen_cpp/olap_file.pb.h>
@@ -66,7 +67,6 @@
 #include "runtime/runtime_predicate.h"
 #include "runtime/runtime_state.h"
 #include "util/coding.h"
-#include "util/crc32c.h"
 #include "util/slice.h" // Slice
 #include "vec/columns/column.h"
 #include "vec/common/schema_util.h"
@@ -469,7 +469,7 @@ Status Segment::_parse_footer(std::shared_ptr<SegmentFooterPB>& footer,
 
     // validate footer PB's checksum
     uint32_t expect_checksum = decode_fixed32_le(fixed_buf + 4);
-    uint32_t actual_checksum = crc32c::Value(footer_buf.data(), footer_buf.size());
+    uint32_t actual_checksum = crc32c::Crc32c(footer_buf.data(), footer_buf.size());
     if (actual_checksum != expect_checksum) {
         Status st = _write_error_file(file_size, file_size - 12 - footer_length, bytes_read,
                                       footer_buf.data(), io_ctx);

be/src/olap/rowset/segment_v2/segment_writer.cpp

@@ -24,6 +24,8 @@
 #include <algorithm>
 
 // IWYU pragma: no_include <opentelemetry/common/threadlocal.h>
+#include <crc32c/crc32c.h>
+
 #include "cloud/config.h"
 #include "common/cast_set.h"
 #include "common/compiler_util.h" // IWYU pragma: keep
@@ -62,7 +64,6 @@
 #include "runtime/memory/mem_tracker.h"
 #include "service/point_query_executor.h"
 #include "util/coding.h"
-#include "util/crc32c.h"
 #include "util/faststring.h"
 #include "util/key_util.h"
 #include "util/simd/bits.h"
@@ -1155,7 +1156,7 @@ Status SegmentWriter::_write_footer() {
     // footer's size
     put_fixed32_le(&fixed_buf, cast_set<uint32_t>(footer_buf.size()));
     // footer's checksum
-    uint32_t checksum = crc32c::Value(footer_buf.data(), footer_buf.size());
+    uint32_t checksum = crc32c::Crc32c(footer_buf.data(), footer_buf.size());
     put_fixed32_le(&fixed_buf, checksum);
     // Append magic number. we don't write magic number in the header because
     // that will need an extra seek when reading

be/src/olap/rowset/segment_v2/vertical_segment_writer.cpp

@@ -17,6 +17,7 @@
 
 #include "olap/rowset/segment_v2/vertical_segment_writer.h"
 
+#include <crc32c/crc32c.h>
 #include <gen_cpp/olap_file.pb.h>
 #include <gen_cpp/segment_v2.pb.h>
 #include <parallel_hashmap/phmap.h>
@@ -63,7 +64,6 @@
 #include "runtime/memory/mem_tracker.h"
 #include "service/point_query_executor.h"
 #include "util/coding.h"
-#include "util/crc32c.h"
 #include "util/debug_points.h"
 #include "util/faststring.h"
 #include "util/key_util.h"
@@ -1371,7 +1371,7 @@ Status VerticalSegmentWriter::_write_footer() {
     // footer's size
     put_fixed32_le(&fixed_buf, cast_set<uint32_t>(footer_buf.size()));
     // footer's checksum
-    uint32_t checksum = crc32c::Value(footer_buf.data(), footer_buf.size());
+    uint32_t checksum = crc32c::Crc32c(footer_buf.data(), footer_buf.size());
     put_fixed32_le(&fixed_buf, checksum);
     // Append magic number. we don't write magic number in the header because
     // that will need an extra seek when reading

be/src/olap/wal/wal_reader.cpp

@@ -17,6 +17,8 @@
 
 #include "olap/wal/wal_reader.h"
 
+#include <crc32c/crc32c.h>
+
 #include <string_view>
 #include <utility>
 
@@ -25,7 +27,6 @@
 #include "io/fs/file_system.h"
 #include "io/fs/path.h"
 #include "util/coding.h"
-#include "util/crc32c.h"
 #include "util/string_util.h"
 #include "wal_writer.h"
 
@@ -145,7 +146,7 @@ Status WalReader::read_header(uint32_t& version, std::string& col_ids) {
 }
 
 Status WalReader::_check_checksum(const char* binary, size_t size, uint32_t checksum) {
-    uint32_t computed_checksum = crc32c::Value(binary, size);
+    uint32_t computed_checksum = crc32c::Crc32c(binary, size);
     if (LIKELY(computed_checksum == checksum)) {
         return Status::OK();
     }