[feature](bucket) Support custom distribution_hash_type for Hash Bucketing

be/src/agent/be_exec_version_manager.cpp

@@ -132,7 +132,10 @@ void BeExecVersionManager::check_function_compatibility(int current_be_exec_vers
 //   a. support strict ownership hash routing for external table sink writers.
 //   b. support Paimon default fixed-bucket routing in the external sink exchange.
 
-const int BeExecVersionManager::max_be_exec_version = 13;
+// 14: start from master
+//   a. support pluggable hash algorithms for table distribution and bucket-local exchanges.
+
+const int BeExecVersionManager::max_be_exec_version = SUPPORT_DISTRIBUTION_HASH_TYPE_VERSION;
 const int BeExecVersionManager::min_be_exec_version = 0;
 std::map<std::string, std::set<int>> BeExecVersionManager::_function_change_map {};
 std::set<std::string> BeExecVersionManager::_function_restrict_map;

be/src/agent/be_exec_version_manager.h

@@ -29,6 +29,7 @@ constexpr inline int USE_NEW_FIXED_OBJECT_SERIALIZATION_VERSION = 10;
 constexpr inline int SUPPORT_ICEBERG_MERGE_CARDINALITY_VERSION = 11;
 constexpr inline int SUPPORT_ICEBERG_VARIANT_VERSION = 12;
 constexpr inline int SUPPORT_EXTERNAL_TABLE_SINK_HASH_VERSION = 13;
+constexpr inline int SUPPORT_DISTRIBUTION_HASH_TYPE_VERSION = 14;
 
 class BeExecVersionManager {
 public:

be/src/exec/exchange/local_exchange_sink_operator.cpp

@@ -50,7 +50,17 @@ Status LocalExchangeSinkOperatorX::_create_partitioner(RuntimeState* state, int
         RETURN_IF_ERROR(_partitioner->init(_texprs));
     } else if (_type == TLocalPartitionType::BUCKET_HASH_SHUFFLE) {
         DCHECK_GT(bucket_count, 0);
-        _partitioner = std::make_unique<Crc32HashPartitioner<ShuffleChannelIds>>(bucket_count);
+        switch (_distribution_hash_type) {
+        case TDistributionHashType::CRC32:
+            _partitioner = std::make_unique<Crc32HashPartitioner<ShuffleChannelIds>>(bucket_count);
+            break;
+        case TDistributionHashType::IDENTITY:
+            _partitioner = std::make_unique<IdentityHashPartitioner>(bucket_count);
+            break;
+        default:
+            return Status::InternalError("unsupported distribution_hash_type {}",
+                                         static_cast<int>(_distribution_hash_type));
+        }
         RETURN_IF_ERROR(_partitioner->init(_texprs));
     }
     return Status::OK();

be/src/exec/exchange/local_exchange_sink_operator.h

@@ -73,8 +73,10 @@ class LocalExchangeSinkOperatorX final : public DataSinkOperatorX<LocalExchangeS
     using Base = DataSinkOperatorX<LocalExchangeSinkLocalState>;
     LocalExchangeSinkOperatorX(int sink_id, int dest_id, int num_partitions,
                                const std::vector<TExpr>& texprs,
-                               const std::map<int, int>& bucket_seq_to_instance_idx)
+                               const std::map<int, int>& bucket_seq_to_instance_idx,
+                               TDistributionHashType::type distribution_hash_type)
             : Base(sink_id, dest_id, dest_id),
+              _distribution_hash_type(distribution_hash_type),
               _num_partitions(num_partitions),
               _texprs(texprs),
               _partitioned_exprs_num(texprs.size()),
@@ -85,6 +87,9 @@ class LocalExchangeSinkOperatorX final : public DataSinkOperatorX<LocalExchangeS
                                const std::map<int, int>& shuffle_id_to_instance_idx)
             : Base(operator_id, tnode, dest_id),
               _type(tnode.local_exchange_node.partition_type),
+              _distribution_hash_type(tnode.local_exchange_node.__isset.distribution_hash_type
+                                              ? tnode.local_exchange_node.distribution_hash_type
+                                              : TDistributionHashType::CRC32),
               _num_partitions(num_partitions),
               _texprs(tnode.local_exchange_node.distribute_expr_lists),
               _partitioned_exprs_num(tnode.local_exchange_node.distribute_expr_lists.size()),
@@ -135,6 +140,7 @@ class LocalExchangeSinkOperatorX final : public DataSinkOperatorX<LocalExchangeS
     Status _create_partitioner(RuntimeState* state, int bucket_count);
 
     TLocalPartitionType::type _type;
+    const TDistributionHashType::type _distribution_hash_type = TDistributionHashType::CRC32;
     const int _num_partitions;
     const std::vector<TExpr>& _texprs;
     const size_t _partitioned_exprs_num;

be/src/exec/operator/exchange_sink_operator.cpp

@@ -138,11 +138,24 @@ Status ExchangeSinkLocalState::init(RuntimeState* state, LocalSinkStateInfo& inf
                 "Partitioner", fmt::format("Crc32CHashPartitioner({})", _partition_count));
     } else if (_part_type == TPartitionType::BUCKET_SHFFULE_HASH_PARTITIONED) {
         _partition_count = channels.size();
-        _partitioner = std::make_unique<Crc32HashPartitioner<ShuffleChannelIds>>(channels.size());
+        switch (p._distribution_hash_type) {
+        case TDistributionHashType::CRC32:
+            _partitioner =
+                    std::make_unique<Crc32HashPartitioner<ShuffleChannelIds>>(channels.size());
+            custom_profile()->add_info_string(
+                    "Partitioner", fmt::format("Crc32HashPartitioner({})", _partition_count));
+            break;
+        case TDistributionHashType::IDENTITY:
+            _partitioner = std::make_unique<IdentityHashPartitioner>(channels.size());
+            custom_profile()->add_info_string(
+                    "Partitioner", fmt::format("IdentityHashPartitioner({})", _partition_count));
+            break;
+        default:
+            return Status::InternalError("unsupported distribution_hash_type {}",
+                                         static_cast<int>(p._distribution_hash_type));
+        }
         RETURN_IF_ERROR(_partitioner->init(p._texprs));
         RETURN_IF_ERROR(_partitioner->prepare(state, p._row_desc));
-        custom_profile()->add_info_string(
-                "Partitioner", fmt::format("Crc32HashPartitioner({})", _partition_count));
     } else if (_part_type == TPartitionType::OLAP_TABLE_SINK_HASH_PARTITIONED) {
         // in ExchangeOlapWriter we rely on type of _partitioner here
         _partition_count = channels.size();
@@ -301,6 +314,9 @@ ExchangeSinkOperatorX::ExchangeSinkOperatorX(
           _texprs(sink.output_partition.partition_exprs),
           _row_desc(row_desc),
           _part_type(sink.output_partition.type),
+          _distribution_hash_type(sink.output_partition.__isset.distribution_hash_type
+                                          ? sink.output_partition.distribution_hash_type
+                                          : TDistributionHashType::CRC32),
           _dests(destinations),
           _dest_node_id(sink.dest_node_id),
           _transfer_large_data_by_brpc(config::transfer_large_data_by_brpc),

be/src/exec/operator/exchange_sink_operator.h

@@ -250,6 +250,7 @@ class ExchangeSinkOperatorX MOCK_REMOVE(final) : public DataSinkOperatorX<Exchan
     TTupleId _output_tuple_id = -1;
 
     TPartitionType::type _part_type;
+    const TDistributionHashType::type _distribution_hash_type = TDistributionHashType::CRC32;
 
     // serialized batches for broadcasting; we need two so we can write
     // one while the other one is still being sent

be/src/exec/partitioner/partitioner.cpp

@@ -23,6 +23,7 @@
 #include "exec/exchange/local_exchange_sink_operator.h"
 #include "exec/exchange/vdata_stream_sender.h"
 #include "runtime/thread_context.h"
+#include "util/raw_value.h"
 
 namespace doris {
 
@@ -84,6 +85,23 @@ Status Crc32CHashPartitioner::clone(RuntimeState* state,
     return _clone_expr_ctxs(state, new_partitioner->_partition_expr_ctxs);
 }
 
+void IdentityHashPartitioner::_do_hash(const ColumnPtr& column, HashValType* __restrict result,
+                                       int idx) const {
+    const PrimitiveType type = _partition_expr_ctxs[idx]->root()->data_type()->get_primitive_type();
+    for (size_t row = 0; row < column->size(); ++row) {
+        auto val = column->get_data_at(row);
+        result[row] =
+                RawValue::identity_hash(val.data, val.size, type, result[row], _partition_count);
+    }
+}
+
+Status IdentityHashPartitioner::clone(RuntimeState* state,
+                                      std::unique_ptr<PartitionerBase>& partitioner) {
+    auto* new_partitioner = new IdentityHashPartitioner(_partition_count);
+    partitioner.reset(new_partitioner);
+    return _clone_expr_ctxs(state, new_partitioner->_partition_expr_ctxs);
+}
+
 template class Crc32HashPartitioner<ShuffleChannelIds>;
 template class Crc32HashPartitioner<SpillPartitionChannelIds>;
 template class Crc32HashPartitioner<SpillRePartitionChannelIds>;

be/src/exec/partitioner/partitioner.h

@@ -191,6 +191,26 @@ class Crc32CHashPartitioner : public Crc32HashPartitioner<ShiftChannelIds> {
     }
 };
 
+// Bucket-shuffle repartitioner for tables bucketed with the identity hash. Each distribution
+// column's canonical bytes are interpreted as an unsigned integer with the first byte as the least
+// significant, then appended to the preceding columns; the combined value is kept modulo the
+// bucket count. Must stay bit-identical with FE HashDistributionPruner and BE tablet routing.
+class IdentityHashPartitioner : public Crc32HashPartitioner<ShuffleChannelIds> {
+public:
+    IdentityHashPartitioner(int partition_count)
+            : Crc32HashPartitioner<ShuffleChannelIds>(partition_count) {}
+
+    Status clone(RuntimeState* state, std::unique_ptr<PartitionerBase>& partitioner) override;
+
+private:
+    void _do_hash(const ColumnPtr& column, HashValType* __restrict result, int idx) const override;
+
+    void _initialize_hash_vals(size_t rows) const override {
+        _hash_vals.resize(rows);
+        std::ranges::fill(_hash_vals, 0);
+    }
+};
+
 /// Instantiated once in partitioner.cpp; suppresses per-TU implicit instantiation.
 extern template class Crc32HashPartitioner<ShuffleChannelIds>;
 extern template class Crc32HashPartitioner<SpillPartitionChannelIds>;

be/src/exec/pipeline/dependency.h

@@ -790,11 +790,17 @@ struct DataDistribution {
     DataDistribution(TLocalPartitionType::type type) : distribution_type(type) {}
     DataDistribution(TLocalPartitionType::type type, const std::vector<TExpr>& partition_exprs_)
             : distribution_type(type), partition_exprs(partition_exprs_) {}
+    DataDistribution(TLocalPartitionType::type type, const std::vector<TExpr>& partition_exprs_,
+                     TDistributionHashType::type distribution_hash_type_)
+            : distribution_type(type),
+              partition_exprs(partition_exprs_),
+              distribution_hash_type(distribution_hash_type_) {}
     DataDistribution(const DataDistribution& other) = default;
     bool need_local_exchange() const { return distribution_type != TLocalPartitionType::NOOP; }
     DataDistribution& operator=(const DataDistribution& other) = default;
     TLocalPartitionType::type distribution_type;
     std::vector<TExpr> partition_exprs;
+    TDistributionHashType::type distribution_hash_type = TDistributionHashType::CRC32;
 };
 
 class ExchangerBase;

be/src/exec/pipeline/pipeline_fragment_context.cpp

@@ -1015,9 +1015,14 @@ Status PipelineFragmentContext::_add_local_exchange_impl(
     const bool use_global_hash_shuffle = bucket_seq_to_instance_idx.empty() &&
                                          !shuffle_idx_to_instance_idx.contains(-1) &&
                                          followed_by_shuffled_operator && !_use_serial_source;
+    if (data_distribution.distribution_type == TLocalPartitionType::BUCKET_HASH_SHUFFLE &&
+        _params.fragment.__isset.distribution_hash_type) {
+        data_distribution.distribution_hash_type = _params.fragment.distribution_hash_type;
+    }
     sink = std::make_shared<LocalExchangeSinkOperatorX>(
             sink_id, local_exchange_id, use_global_hash_shuffle ? _total_instances : _num_instances,
-            data_distribution.partition_exprs, bucket_seq_to_instance_idx);
+            data_distribution.partition_exprs, bucket_seq_to_instance_idx,
+            data_distribution.distribution_hash_type);
     if (bucket_seq_to_instance_idx.empty() &&
         data_distribution.distribution_type == TLocalPartitionType::BUCKET_HASH_SHUFFLE) {
         data_distribution.distribution_type =

be/src/storage/tablet_info.cpp

@@ -29,6 +29,7 @@
 #include <algorithm>
 #include <cstddef>
 #include <cstdint>
+#include <cstring>
 #include <memory>
 #include <ostream>
 #include <string>

be/src/storage/tablet_info.h

@@ -248,24 +248,40 @@ class VOlapTablePartitionParam {
             std::map<VOlapTablePartition*, int64_t>* partition_tablets_buffer = nullptr) const {
         std::function<uint32_t(Block*, uint32_t, const VOlapTablePartition&)> compute_function;
         if (!_distributed_slot_locs.empty()) {
-            //TODO: refactor by saving the hash values. then we can calculate in columnwise.
-            compute_function = [this](Block* block, uint32_t row,
-                                      const VOlapTablePartition& partition) -> uint32_t {
-                uint32_t hash_val = 0;
-                for (unsigned short _distributed_slot_loc : _distributed_slot_locs) {
-                    auto* slot_desc = _slots[_distributed_slot_loc];
-                    auto& column = block->get_by_position(_distributed_slot_loc).column;
-                    auto val = column->get_data_at(row);
-                    if (val.data != nullptr) {
-                        hash_val = RawValue::zlib_crc32(val.data, val.size,
-                                                        slot_desc->type()->get_primitive_type(),
-                                                        hash_val);
-                    } else {
-                        hash_val = HashUtil::zlib_crc_hash_null(hash_val);
+            if (_t_param.distribution_hash_type == TDistributionHashType::IDENTITY) {
+                compute_function = [this](Block* block, uint32_t row,
+                                          const VOlapTablePartition& partition) -> uint32_t {
+                    uint32_t bucket = 0;
+                    for (unsigned short distributed_slot_loc : _distributed_slot_locs) {
+                        auto* slot_desc = _slots[distributed_slot_loc];
+                        const auto& column = block->get_by_position(distributed_slot_loc).column;
+                        auto val = column->get_data_at(row);
+                        bucket = RawValue::identity_hash(
+                                val.data, val.size, slot_desc->type()->get_primitive_type(), bucket,
+                                cast_set<uint32_t>(partition.num_buckets));
                     }
-                }
-                return cast_set<uint32_t>(hash_val % partition.num_buckets);
-            };
+                    return bucket;
+                };
+            } else {
+                //TODO: refactor by saving the hash values. then we can calculate in columnwise.
+                compute_function = [this](Block* block, uint32_t row,
+                                          const VOlapTablePartition& partition) -> uint32_t {
+                    uint32_t hash_val = 0;
+                    for (unsigned short _distributed_slot_loc : _distributed_slot_locs) {
+                        auto* slot_desc = _slots[_distributed_slot_loc];
+                        auto& column = block->get_by_position(_distributed_slot_loc).column;
+                        auto val = column->get_data_at(row);
+                        if (val.data != nullptr) {
+                            hash_val = RawValue::zlib_crc32(val.data, val.size,
+                                                            slot_desc->type()->get_primitive_type(),
+                                                            hash_val);
+                        } else {
+                            hash_val = HashUtil::zlib_crc_hash_null(hash_val);
+                        }
+                    }
+                    return cast_set<uint32_t>(hash_val % partition.num_buckets);
+                };
+            }
         } else { // random distribution
             compute_function = [](Block* block, uint32_t row,
                                   const VOlapTablePartition& partition) -> uint32_t {

be/src/util/raw_value.h

@@ -22,6 +22,7 @@
 
 #include <string>
 
+#include "common/check.h"
 #include "common/consts.h"
 #include "common/logging.h"
 #include "core/data_type/define_primitive_type.h"
@@ -38,8 +39,98 @@ class RawValue {
     // Same as the up function, only use in vec exec engine.
     static uint32_t zlib_crc32(const void* value, size_t len, const PrimitiveType& type,
                                uint32_t seed);
+
+    // Treat the canonical distribution bytes of a value as an unsigned integer with the first byte
+    // as the least-significant byte, then append it to the preceding distribution columns. The
+    // returned value is kept modulo mod throughout, so values of any byte width and any number of
+    // columns do not require a wide integer.
+    static uint32_t identity_hash(const void* value, size_t len, const PrimitiveType& type,
+                                  uint32_t seed, uint32_t mod);
 };
 
+inline uint32_t RawValue::identity_hash(const void* v, size_t len, const PrimitiveType& type,
+                                        uint32_t seed, uint32_t mod) {
+    DCHECK_GT(mod, 0);
+    auto append_little_endian = [&seed, mod](const void* value, size_t size) {
+        const auto* bytes = reinterpret_cast<const uint8_t*>(value);
+        uint64_t remainder = seed;
+        size_t bytes_since_mod = 0;
+        for (size_t i = size; i > 0; --i) {
+            remainder = remainder * 256 + bytes[i - 1];
+            if (++bytes_since_mod == sizeof(uint32_t)) {
+                remainder %= mod;
+                bytes_since_mod = 0;
+            }
+        }
+        seed = static_cast<uint32_t>(remainder % mod);
+    };
+
+    if (v == nullptr) {
+        static constexpr uint32_t NULL_VALUE = 0;
+        append_little_endian(&NULL_VALUE, sizeof(NULL_VALUE));
+        return seed;
+    }
+
+    switch (type) {
+    case TYPE_VARCHAR:
+    case TYPE_VARBINARY:
+    case TYPE_HLL:
+    case TYPE_STRING:
+    case TYPE_CHAR:
+        append_little_endian(v, len);
+        break;
+    case TYPE_BOOLEAN:
+    case TYPE_TINYINT:
+        append_little_endian(v, 1);
+        break;
+    case TYPE_SMALLINT:
+        append_little_endian(v, 2);
+        break;
+    case TYPE_INT:
+    case TYPE_FLOAT:
+    case TYPE_DATEV2:
+    case TYPE_DECIMAL32:
+    case TYPE_IPV4:
+        append_little_endian(v, 4);
+        break;
+    case TYPE_BIGINT:
+    case TYPE_DOUBLE:
+    case TYPE_TIMEV2:
+    case TYPE_DATETIMEV2:
+    case TYPE_TIMESTAMPTZ:
+    case TYPE_DECIMAL64:
+        append_little_endian(v, 8);
+        break;
+    case TYPE_LARGEINT:
+    case TYPE_DECIMAL128I:
+    case TYPE_IPV6:
+        append_little_endian(v, 16);
+        break;
+    case TYPE_DECIMAL256:
+        append_little_endian(v, 32);
+        break;
+    case TYPE_DATE:
+    case TYPE_DATETIME: {
+        const auto* date_val = reinterpret_cast<const VecDateTimeValue*>(v);
+        char buf[64];
+        int date_len = date_val->to_buffer(buf);
+        append_little_endian(buf, date_len);
+        break;
+    }
+    case TYPE_DECIMALV2: {
+        const auto* dec_val = reinterpret_cast<const DecimalV2Value*>(v);
+        int64_t int_val = dec_val->int_value();
+        int32_t frac_val = dec_val->frac_value();
+        append_little_endian(&frac_val, sizeof(frac_val));
+        append_little_endian(&int_val, sizeof(int_val));
+        break;
+    }
+    default:
+        DORIS_CHECK(false) << "invalid type: " << type;
+    }
+    return seed;
+}
+
 // NOTE: this is just for split data, decimal use old doris hash function
 // Because crc32 hardware is not equal with zlib crc32
 inline uint32_t RawValue::zlib_crc32(const void* v, size_t len, const PrimitiveType& type,
@@ -75,7 +166,7 @@ inline uint32_t RawValue::zlib_crc32(const void* v, size_t len, const PrimitiveT
         return HashUtil::zlib_crc_hash(v, 8, seed);
     case TYPE_DATE:
     case TYPE_DATETIME: {
-        auto* date_val = (const VecDateTimeValue*)v;
+        const auto* date_val = reinterpret_cast<const VecDateTimeValue*>(v);
         char buf[64];
         int date_len = date_val->to_buffer(buf);
         return HashUtil::zlib_crc_hash(buf, date_len, seed);
@@ -94,7 +185,7 @@ inline uint32_t RawValue::zlib_crc32(const void* v, size_t len, const PrimitiveT
     }
 
     case TYPE_DECIMALV2: {
-        const DecimalV2Value* dec_val = (const DecimalV2Value*)v;
+        const auto* dec_val = reinterpret_cast<const DecimalV2Value*>(v);
         int64_t int_val = dec_val->int_value();
         int32_t frac_val = dec_val->frac_value();
         seed = HashUtil::zlib_crc_hash(&int_val, sizeof(int_val), seed);