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

be/src/agent/be_exec_version_manager.cpp

@@ -135,8 +135,10 @@ void BeExecVersionManager::check_function_compatibility(int current_be_exec_vers
 //   a. support TIMESTAMP_NS in Thrift descriptors and PBlock exchange.
 // 15: start from master
 //   a. distinguish Hive OpenCSVSerde row semantics from generic CSV decoding during upgrades.
+// 16: start from master
+//   a. support pluggable hash algorithms for table distribution and bucket-local exchanges.
 
-const int BeExecVersionManager::max_be_exec_version = SUPPORT_HIVE_OPEN_CSV_VERSION;
+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

@@ -35,6 +35,7 @@ constexpr inline int SUPPORT_ICEBERG_VARIANT_VERSION = 12;
 constexpr inline int SUPPORT_EXTERNAL_TABLE_SINK_HASH_VERSION = 13;
 constexpr inline int SUPPORT_TIMESTAMP_NS_VERSION = 14;
 constexpr inline int SUPPORT_HIVE_OPEN_CSV_VERSION = 15;
+constexpr inline int SUPPORT_DISTRIBUTION_HASH_TYPE_VERSION = 16;
 
 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()),
@@ -124,6 +129,10 @@ class LocalExchangeSinkOperatorX final : public DataSinkOperatorX<LocalExchangeS
         local_state._exchanger->set_low_memory_mode();
     }
 
+#ifdef BE_TEST
+    PartitionerBase* partitioner_for_test() const { return _partitioner.get(); }
+#endif
+
 private:
     friend class LocalExchangeSinkLocalState;
     friend class ShuffleExchanger;
@@ -135,6 +144,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

@@ -794,8 +794,13 @@ struct DataDistribution {
     DataDistribution(const DataDistribution& other) = default;
     bool need_local_exchange() const { return distribution_type != TLocalPartitionType::NOOP; }
     DataDistribution& operator=(const DataDistribution& other) = default;
+    // Hash type is fragment-scoped by design: PipelineFragmentContext::_add_local_exchange_impl
+    // stamps the fragment's distribution_hash_type onto every BUCKET_HASH_SHUFFLE local exchange
+    // (FE derives it from the fragment root and rejects mixed-layout fragments before sending),
+    // so per-operator construction never needs to carry one.
     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

@@ -1017,9 +1017,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/exec/runtime_filter/runtime_filter_bucket_pruner.cpp

@@ -23,6 +23,7 @@
 #include <memory>
 #include <mutex>
 
+#include "common/cast_set.h"
 #include "exprs/hybrid_set.h"
 #include "exprs/runtime_filter_expr.h"
 #include "exprs/vexpr.h"
@@ -40,14 +41,21 @@ Status RuntimeFilterBucketPruner::prune_by_runtime_filters(
         return Status::OK();
     }
 
-    phmap::flat_hash_set<int> eligible_filter_ids;
+    phmap::flat_hash_map<int, TDistributionHashType::type> eligible_filter_hash_types;
     for (const auto& desc : rf_descs) {
         if (desc.__isset.bucket_pruning_target_ids &&
             desc.bucket_pruning_target_ids.contains(scan_node_id)) {
-            eligible_filter_ids.insert(desc.filter_id);
+            TDistributionHashType::type hash_type = TDistributionHashType::CRC32;
+            if (desc.__isset.bucket_pruning_target_hash_types) {
+                auto it = desc.bucket_pruning_target_hash_types.find(scan_node_id);
+                if (it != desc.bucket_pruning_target_hash_types.end()) {
+                    hash_type = it->second;
+                }
+            }
+            eligible_filter_hash_types.emplace(desc.filter_id, hash_type);
         }
     }
-    if (eligible_filter_ids.empty()) {
+    if (eligible_filter_hash_types.empty()) {
         return Status::OK();
     }
 
@@ -57,7 +65,8 @@ Status RuntimeFilterBucketPruner::prune_by_runtime_filters(
             continue;
         }
         auto* rf_expr = assert_cast<RuntimeFilterExpr*>(root.get());
-        if (!eligible_filter_ids.contains(rf_expr->filter_id())) {
+        auto hash_type_it = eligible_filter_hash_types.find(rf_expr->filter_id());
+        if (hash_type_it == eligible_filter_hash_types.end()) {
             continue;
         }
 
@@ -77,8 +86,6 @@ Status RuntimeFilterBucketPruner::prune_by_runtime_filters(
         VExprSPtr target_expr = impl->children()[0];
         DORIS_CHECK_EQ(target_expr->node_type(), TExprNodeType::SLOT_REF);
 
-        std::shared_ptr<const std::vector<uint32_t>> hashes =
-                rf_expr->get_bucket_prune_hashes(target_expr->data_type());
         phmap::flat_hash_map<int32_t, phmap::flat_hash_set<int32_t>> new_selected_buckets_by_num;
         for (const auto& range_ptr : ranges) {
             DORIS_CHECK(range_ptr != nullptr);
@@ -92,6 +99,10 @@ Status RuntimeFilterBucketPruner::prune_by_runtime_filters(
             auto [selected_it, inserted] =
                     new_selected_buckets_by_num.try_emplace(range.bucket_num);
             if (inserted) {
+                std::shared_ptr<const std::vector<uint32_t>> hashes =
+                        rf_expr->get_bucket_prune_hashes(target_expr->data_type(),
+                                                         hash_type_it->second,
+                                                         cast_set<uint32_t>(range.bucket_num));
                 auto& selected_buckets = selected_it->second;
                 selected_buckets.reserve(
                         std::min(hashes->size(), static_cast<size_t>(range.bucket_num)));

be/src/exec/runtime_filter/runtime_filter_wrapper.cpp

@@ -17,12 +17,16 @@
 
 #include "exec/runtime_filter/runtime_filter_wrapper.h"
 
+#include <algorithm>
+
 #include "core/data_type/define_primitive_type.h"
 #include "core/string_ref.h"
+#include "exec/common/hash_table/phmap_fwd_decl.h"
 #include "exec/runtime_filter/runtime_filter_definitions.h"
 #include "exprs/create_predicate_function.h"
 #include "exprs/function/cast/cast_to_date_or_datetime_impl.hpp"
 #include "util/hash_util.hpp"
+#include "util/raw_value.h"
 
 namespace doris {
 RuntimeFilterWrapper::RuntimeFilterWrapper(const RuntimeFilterParams* params)
@@ -632,14 +636,60 @@ bool RuntimeFilterWrapper::contain_null() const {
     return false;
 }
 
+std::shared_ptr<const std::vector<uint32_t>> RuntimeFilterWrapper::_get_or_compute_identity_buckets(
+        PrimitiveType primitive_type, uint32_t bucket_num) const {
+    std::scoped_lock lock(_identity_bucket_prune_hashes_mutex);
+    if (auto it = _identity_bucket_prune_hashes.find(bucket_num);
+        it != _identity_bucket_prune_hashes.end()) {
+        return it->second;
+    }
+    _bucket_prune_hashes_started.store(true);
+    // Cache bucket membership, not one entry per IN value: different partitions can use
+    // different bucket counts, and retaining every value would multiply the set size by
+    // the number of counts. Once all buckets are selected, membership cannot change.
+    flat_hash_set<uint32_t> selected_buckets;
+    selected_buckets.reserve(std::min(
+            static_cast<size_t>(bucket_num),
+            static_cast<size_t>(_hybrid_set->size()) + (_hybrid_set->contain_null() ? 1 : 0)));
+    if (_hybrid_set->contain_null()) {
+        selected_buckets.insert(RawValue::identity_hash(nullptr, 0, primitive_type, 0, bucket_num));
+    }
+    auto* iter = _hybrid_set->begin();
+    while (selected_buckets.size() < bucket_num && iter->has_next()) {
+        const void* value = iter->get_value();
+        DORIS_CHECK(value != nullptr);
+        if (is_string_type(primitive_type) || primitive_type == TYPE_VARBINARY) {
+            const auto* string_value = reinterpret_cast<const StringRef*>(value);
+            selected_buckets.insert(RawValue::identity_hash(string_value->data, string_value->size,
+                                                            primitive_type, 0, bucket_num));
+        } else {
+            selected_buckets.insert(
+                    RawValue::identity_hash(value, 0, primitive_type, 0, bucket_num));
+        }
+        iter->next();
+    }
+    auto buckets = std::make_shared<std::vector<uint32_t>>(selected_buckets.begin(),
+                                                           selected_buckets.end());
+    _identity_bucket_prune_hashes.emplace(bucket_num, buckets);
+    return buckets;
+}
+
 std::shared_ptr<const std::vector<uint32_t>>
-RuntimeFilterWrapper::get_or_compute_bucket_prune_hashes(const DataTypePtr& target_type) const {
+RuntimeFilterWrapper::get_or_compute_bucket_prune_hashes(const DataTypePtr& target_type,
+                                                         TDistributionHashType::type hash_type,
+                                                         uint32_t bucket_num) const {
     DORIS_CHECK(_state.load() == State::READY);
     DORIS_CHECK(_hybrid_set != nullptr);
     DORIS_CHECK(target_type != nullptr);
+    DORIS_CHECK_GT(bucket_num, 0);
     PrimitiveType primitive_type = target_type->get_primitive_type();
     DORIS_CHECK_EQ(primitive_type, _column_return_type);
 
+    if (hash_type == TDistributionHashType::IDENTITY) {
+        return _get_or_compute_identity_buckets(primitive_type, bucket_num);
+    }
+
+    DORIS_CHECK_EQ(hash_type, TDistributionHashType::CRC32);
     std::call_once(_bucket_prune_hashes_once, [&] {
         _bucket_prune_hashes_started.store(true);
         // Materialize the exact-set values into a column so bucket pruning uses the