[fix](be) Hash VARBINARY values in shuffle and partition exchanges

be/src/agent/be_exec_version_manager.cpp

@@ -158,8 +158,10 @@ void BeExecVersionManager::check_function_compatibility(int current_be_exec_vers
 //   a. use exact logical frames for map_agg_v2 aggregate state serialization.
 // 18: start from master
 //   a. support Iceberg partition transform routing in the external sink exchange.
+// 19: start from master
+//   a. support hashing VARBINARY values in a shuffle or partition exchange.
 
-const int BeExecVersionManager::max_be_exec_version = SUPPORT_ICEBERG_TRANSFORM_ROUTING_VERSION;
+const int BeExecVersionManager::max_be_exec_version = SUPPORT_VARBINARY_SHUFFLE_HASH_VERSION;
 const int BeExecVersionManager::min_be_exec_version = 0;
 std::map<std::string, std::set<int>> BeExecVersionManager::_function_change_map {};
 std::map<std::string, std::set<int>> BeExecVersionManager::_function_alternative_map {};

be/src/agent/be_exec_version_manager.h

@@ -40,6 +40,7 @@ constexpr inline int SUPPORT_HIVE_OPEN_CSV_VERSION = 15;
 constexpr inline int SUPPORT_PAIMON_WRITE_VERSION = 16;
 constexpr inline int SUPPORT_MAP_AGG_V2_EXACT_FRAME_VERSION = 17;
 constexpr inline int SUPPORT_ICEBERG_TRANSFORM_ROUTING_VERSION = 18;
+constexpr inline int SUPPORT_VARBINARY_SHUFFLE_HASH_VERSION = 19;
 
 class BeExecVersionManager {
 public:

be/src/core/column/column_varbinary.cpp

@@ -17,6 +17,7 @@
 
 #include "core/column/column_varbinary.h"
 
+#include <crc32c/crc32c.h>
 #include <glog/logging.h>
 
 #include <cstddef>
@@ -29,6 +30,7 @@
 #include "core/column/columns_common.h"
 #include "core/data_type/primitive_type.h"
 #include "exec/sort/sort_block.h"
+#include "util/hash_util.hpp"
 
 namespace doris {
 
@@ -192,6 +194,74 @@ size_t ColumnVarbinary::get_max_row_byte_size() const {
     return max_size + sizeof(uint32_t);
 }
 
+void ColumnVarbinary::update_crcs_with_value(uint32_t* __restrict hashes, PrimitiveType /*type*/,
+                                             uint32_t rows, uint32_t /*offset*/,
+                                             const uint8_t* __restrict null_data) const {
+    DCHECK(rows == size());
+    if (null_data == nullptr) {
+        for (size_t i = 0; i < rows; ++i) {
+            hashes[i] = HashUtil::zlib_crc_hash(_data[i].data(), _data[i].size(), hashes[i]);
+        }
+    } else {
+        for (size_t i = 0; i < rows; ++i) {
+            if (null_data[i] == 0) {
+                hashes[i] = HashUtil::zlib_crc_hash(_data[i].data(), _data[i].size(), hashes[i]);
+            }
+        }
+    }
+}
+
+void ColumnVarbinary::update_crc_with_value(size_t start, size_t end, uint32_t& hash,
+                                            const uint8_t* __restrict null_data) const {
+    if (null_data == nullptr) {
+        for (size_t i = start; i < end; ++i) {
+            hash = HashUtil::zlib_crc_hash(_data[i].data(), _data[i].size(), hash);
+        }
+    } else {
+        for (size_t i = start; i < end; ++i) {
+            if (null_data[i] == 0) {
+                hash = HashUtil::zlib_crc_hash(_data[i].data(), _data[i].size(), hash);
+            }
+        }
+    }
+}
+
+void ColumnVarbinary::update_crc32c_batch(uint32_t* __restrict hashes,
+                                          const uint8_t* __restrict null_map) const {
+    const size_t rows = size();
+    if (null_map == nullptr) {
+        for (size_t i = 0; i < rows; ++i) {
+            hashes[i] = crc32c_extend(hashes[i], reinterpret_cast<const uint8_t*>(_data[i].data()),
+                                      _data[i].size());
+        }
+    } else {
+        for (size_t i = 0; i < rows; ++i) {
+            if (null_map[i] == 0) {
+                hashes[i] =
+                        crc32c_extend(hashes[i], reinterpret_cast<const uint8_t*>(_data[i].data()),
+                                      _data[i].size());
+            }
+        }
+    }
+}
+
+void ColumnVarbinary::update_crc32c_single(size_t start, size_t end, uint32_t& hash,
+                                           const uint8_t* __restrict null_map) const {
+    if (null_map == nullptr) {
+        for (size_t i = start; i < end; ++i) {
+            hash = crc32c_extend(hash, reinterpret_cast<const uint8_t*>(_data[i].data()),
+                                 _data[i].size());
+        }
+    } else {
+        for (size_t i = start; i < end; ++i) {
+            if (null_map[i] == 0) {
+                hash = crc32c_extend(hash, reinterpret_cast<const uint8_t*>(_data[i].data()),
+                                     _data[i].size());
+            }
+        }
+    }
+}
+
 void ColumnVarbinary::serialize(StringRef* keys, size_t num_rows) const {
     for (size_t i = 0; i < num_rows; ++i) {
         keys[i].size += serialize_impl(const_cast<char*>(keys[i].data + keys[i].size), i);

be/src/core/column/column_varbinary.h

@@ -30,7 +30,10 @@
 #include "core/string_view.h"
 
 namespace doris {
-// Binary IO does not enable hash computation; inherit IColumn's unsupported methods.
+// An exchange that partitions rows by a VARBINARY value, such as a shuffle join on ARRAY<VARBINARY>
+// keys or an Iceberg write routed by identity(binary_col), hashes the value's bytes with the
+// CRC32 / CRC32C methods, exactly as ColumnString does: X'616263' and 'abc' get the same hash.
+// The value hashes (SipHash, xxHash) are inherited from IColumn and stay unsupported.
 class ColumnVarbinary final : public COWHelper<IColumn, ColumnVarbinary> {
 private:
     using Self = ColumnVarbinary;
@@ -123,6 +126,19 @@ class ColumnVarbinary final : public COWHelper<IColumn, ColumnVarbinary> {
 
     size_t get_max_row_byte_size() const override;
 
+    void update_crcs_with_value(uint32_t* __restrict hashes, PrimitiveType type, uint32_t rows,
+                                uint32_t offset,
+                                const uint8_t* __restrict null_data) const override;
+
+    void update_crc_with_value(size_t start, size_t end, uint32_t& hash,
+                               const uint8_t* __restrict null_data) const override;
+
+    void update_crc32c_batch(uint32_t* __restrict hashes,
+                             const uint8_t* __restrict null_map) const override;
+
+    void update_crc32c_single(size_t start, size_t end, uint32_t& hash,
+                              const uint8_t* __restrict null_map) const override;
+
     void deserialize(StringRef* keys, const size_t num_rows) override;
 
     void serialize(StringRef* keys, const size_t num_rows) const override;

be/test/core/column/column_varbinary_test.cpp

@@ -26,12 +26,16 @@
 #include <cstring>
 #include <iostream>
 #include <string>
+#include <utility>
 #include <vector>
 
 #include "common/exception.h"
 #include "core/assert_cast.h"
 #include "core/column/column.h"
+#include "core/column/column_array.h"
+#include "core/column/column_nullable.h"
 #include "core/column/column_string.h"
+#include "core/column/column_vector.h"
 #include "core/data_type/primitive_type.h"
 #include "core/string_ref.h"
 #include "core/string_view.h"
@@ -250,21 +254,113 @@ TEST_F(ColumnVarbinaryTest, TabletRoutingHashIsNotSupported) {
     }
 }
 
-TEST_F(ColumnVarbinaryTest, HashingIsNotSupported) {
+TEST_F(ColumnVarbinaryTest, ShuffleHashesMatchStringOfTheSameBytes) {
+    // An exchange routes a VARBINARY value by its bytes, as it routes a STRING. Empty, NUL and
+    // invalid UTF-8 values, and values on both sides of the inline size, must hash like the same
+    // bytes in a ColumnString.
+    const std::vector<std::string> values = {"",
+                                             std::string("\0\xff", 2),
+                                             "abc",
+                                             make_bytes(doris::StringView::kInlineSize, 0x11),
+                                             make_bytes(doris::StringView::kInlineSize + 1, 0x22),
+                                             make_bytes(40, 0x80)};
+    const std::vector<uint8_t> null_map = {0, 1, 0, 0, 1, 0};
+    auto binary = ColumnVarbinary::create();
+    auto string = ColumnString::create();
+    for (const auto& value : values) {
+        binary->insert_data(value.data(), value.size());
+        string->insert_data(value.data(), value.size());
+    }
+    const auto rows = static_cast<uint32_t>(values.size());
+    // Distinct seeds, so a method that ignored the bytes or the row would not match by accident.
+    auto seeds = [&]() {
+        std::vector<uint32_t> hashes(rows);
+        for (uint32_t i = 0; i < rows; ++i) {
+            hashes[i] = 7 * i + 1;
+        }
+        return hashes;
+    };
+
+    for (const uint8_t* nulls : {static_cast<const uint8_t*>(nullptr), null_map.data()}) {
+        SCOPED_TRACE(nulls == nullptr ? "without null map" : "with null map");
+        auto binary_crcs = seeds();
+        auto string_crcs = seeds();
+        binary->update_crcs_with_value(binary_crcs.data(), TYPE_VARBINARY, rows, 0, nulls);
+        string->update_crcs_with_value(string_crcs.data(), TYPE_STRING, rows, 0, nulls);
+        EXPECT_EQ(binary_crcs, string_crcs);
+        EXPECT_NE(binary_crcs, seeds());
+
+        auto binary_crc32cs = seeds();
+        auto string_crc32cs = seeds();
+        binary->update_crc32c_batch(binary_crc32cs.data(), nulls);
+        string->update_crc32c_batch(string_crc32cs.data(), nulls);
+        EXPECT_EQ(binary_crc32cs, string_crc32cs);
+        EXPECT_NE(binary_crc32cs, seeds());
+
+        for (const auto& [start, end] :
+             std::vector<std::pair<size_t, size_t>> {{0, rows}, {1, rows - 1}}) {
+            uint32_t binary_crc = 23;
+            uint32_t string_crc = 23;
+            binary->update_crc_with_value(start, end, binary_crc, nulls);
+            string->update_crc_with_value(start, end, string_crc, nulls);
+            EXPECT_EQ(binary_crc, string_crc);
+            EXPECT_NE(binary_crc, 23);
+
+            uint32_t binary_crc32c = 29;
+            uint32_t string_crc32c = 29;
+            binary->update_crc32c_single(start, end, binary_crc32c, nulls);
+            string->update_crc32c_single(start, end, string_crc32c, nulls);
+            EXPECT_EQ(binary_crc32c, string_crc32c);
+            EXPECT_NE(binary_crc32c, 29);
+        }
+    }
+}
+
+TEST_F(ColumnVarbinaryTest, ArrayOfVarbinaryHashesLikeArrayOfString) {
+    // A shuffle join on ARRAY<VARBINARY> keys hashes the array elements through
+    // update_crc32c_single.
+    // Rows: [X'01', X'0203'], [], [NULL, X'01'].
+    auto make_array = [](MutableColumnPtr nested) {
+        nested->insert_data("\x01", 1);
+        nested->insert_data("\x02\x03", 2);
+        nested->insert_default();
+        nested->insert_data("\x01", 1);
+        auto null_map = ColumnUInt8::create();
+        for (uint8_t is_null : {0, 0, 1, 0}) {
+            null_map->insert_value(is_null);
+        }
+        auto offsets = ColumnArray::ColumnOffsets::create();
+        for (uint64_t offset : {2, 2, 4}) {
+            offsets->insert_value(offset);
+        }
+        return ColumnArray::create(ColumnNullable::create(std::move(nested), std::move(null_map)),
+                                   std::move(offsets));
+    };
+    auto binary = make_array(ColumnVarbinary::create());
+    auto string = make_array(ColumnString::create());
+
+    std::vector<uint32_t> binary_crc32cs(3, 5);
+    std::vector<uint32_t> string_crc32cs(3, 5);
+    binary->update_crc32c_batch(binary_crc32cs.data(), nullptr);
+    string->update_crc32c_batch(string_crc32cs.data(), nullptr);
+    EXPECT_EQ(binary_crc32cs, string_crc32cs);
+
+    std::vector<uint32_t> binary_crcs(3, 5);
+    std::vector<uint32_t> string_crcs(3, 5);
+    binary->update_crcs_with_value(binary_crcs.data(), TYPE_ARRAY, 3, 0, nullptr);
+    string->update_crcs_with_value(string_crcs.data(), TYPE_ARRAY, 3, 0, nullptr);
+    EXPECT_EQ(binary_crcs, string_crcs);
+}
+
+TEST_F(ColumnVarbinaryTest, ValueHashesAreNotSupported) {
     auto binary = ColumnVarbinary::create();
     binary->insert_data("\0\xff", 2);
     SipHash sip;
     uint64_t hash64 = 17;
-    uint32_t hash32 = 23;
     EXPECT_THROW(binary->update_hash_with_value(0, sip), Exception);
     EXPECT_THROW(binary->update_hashes_with_value(&hash64, nullptr), Exception);
     EXPECT_THROW(binary->update_xxHash_with_value(0, 1, hash64, nullptr), Exception);
-    EXPECT_THROW(binary->update_crcs_with_value(&hash32, TYPE_VARBINARY, 1, 0, nullptr), Exception);
-    EXPECT_THROW(binary->update_crc_with_value(0, 1, hash32, nullptr), Exception);
-    EXPECT_THROW(binary->update_crc32c_batch(&hash32, nullptr), Exception);
-    EXPECT_THROW(binary->update_crc32c_single(0, 1, hash32, nullptr), Exception);
     EXPECT_EQ(hash64, 17);
-    EXPECT_EQ(hash32, 23);
 }
 
 TEST_F(ColumnVarbinaryTest, InsertFromAndRanges) {

be/test/exec/partitioner/external_table_sink_hash_partitioner_test.cpp

@@ -29,8 +29,10 @@
 
 #include "common/object_pool.h"
 #include "core/block/block.h"
+#include "core/column/column_varbinary.h"
 #include "core/column/column_vector.h"
 #include "core/data_type/data_type_number.h"
+#include "core/data_type/data_type_varbinary.h"
 #include "exec/partitioner/external/paimon_fixed_bucket_partition_function.h"
 #include "runtime/descriptor_helper.h"
 #include "runtime/descriptors.h"
@@ -40,11 +42,13 @@ namespace doris {
 
 class ExternalTableSinkHashPartitionerTest : public testing::Test {
 protected:
+    virtual PrimitiveType key_type() const { return TYPE_INT; }
+
     void SetUp() override {
         TDescriptorTableBuilder table_builder;
         TTupleDescriptorBuilder tuple_builder;
         tuple_builder.add_slot(TSlotDescriptorBuilder()
-                                       .type(TYPE_INT)
+                                       .type(key_type())
                                        .nullable(false)
                                        .column_name("key")
                                        .column_pos(1)
@@ -65,7 +69,7 @@ class ExternalTableSinkHashPartitionerTest : public testing::Test {
         TExprNode node;
         node.__set_node_type(TExprNodeType::SLOT_REF);
         node.__set_num_children(0);
-        node.__set_type(create_type_desc(TYPE_INT));
+        node.__set_type(create_type_desc(key_type()));
         node.__set_is_nullable(false);
 
         TSlotRef slot_ref;
@@ -236,6 +240,70 @@ TEST_F(ExternalTableSinkHashPartitionerTest, IcebergUnsupportedTransformFailsClo
               std::string::npos);
 }
 
+class ExternalTableSinkBinaryHashPartitionerTest : public ExternalTableSinkHashPartitionerTest {
+protected:
+    PrimitiveType key_type() const override { return TYPE_VARBINARY; }
+
+    Block binary_block(const std::vector<std::string>& values) const {
+        auto column = ColumnVarbinary::create();
+        for (const auto& value : values) {
+            column->insert_data(value.data(), value.size());
+        }
+        Block block;
+        block.insert(ColumnWithTypeAndName(std::move(column), std::make_shared<DataTypeVarbinary>(),
+                                           "key"));
+        return block;
+    }
+
+    std::vector<PartitionerBase::HashValType> channels(const std::string& transform,
+                                                       ShuffleHashMethod method,
+                                                       const std::vector<std::string>& values) {
+        TExternalTableSinkHashPartitionInfo info;
+        info.__set_partition_function("iceberg_transform");
+        info.__set_partition_function_options({{"transforms", transform}});
+        info.__set_writer_assignment(TExternalTableSinkWriterAssignment::IDENTITY);
+        ExternalTableSinkHashPartitioner partitioner(64, method, info);
+        EXPECT_TRUE(partitioner.init({slot_ref()}).ok());
+        EXPECT_TRUE(partitioner.prepare(&_state, *_row_descriptor).ok());
+        EXPECT_TRUE(partitioner.open(&_state).ok());
+        Block input = binary_block(values);
+        Status status = partitioner.do_partitioning(&_state, &input);
+        EXPECT_TRUE(status.ok()) << status.to_string();
+        std::vector<PartitionerBase::HashValType> result(partitioner.get_channel_ids().begin(),
+                                                         partitioner.get_channel_ids().end());
+        EXPECT_TRUE(partitioner.close(&_state).ok());
+        return result;
+    }
+};
+
+TEST_F(ExternalTableSinkBinaryHashPartitionerTest, IcebergTransformHashesBinaryPartitionValues) {
+    // identity(binary_key) and truncate(1, binary_key) route each row by a VARBINARY partition
+    // value, which the exchange hashes by its bytes. A backend without VARBINARY shuffle hashes
+    // failed the write with "update_crcs_with_value is not supported for ColumnVarbinary".
+    const std::string long_value(20, '\x7f');
+    for (ShuffleHashMethod method : {ShuffleHashMethod::CRC32, ShuffleHashMethod::CRC32C}) {
+        SCOPED_TRACE(method == ShuffleHashMethod::CRC32 ? "CRC32" : "CRC32C");
+        // identity: equal bytes share a channel, including the empty value and a value past the
+        // inline size.
+        auto identity = channels("identity", method,
+                                 {std::string("\x00\xff", 2), "", long_value,
+                                  std::string("\x00\xff", 2), "", long_value});
+        ASSERT_EQ(6, identity.size());
+        EXPECT_EQ(identity[0], identity[3]);
+        EXPECT_EQ(identity[1], identity[4]);
+        EXPECT_EQ(identity[2], identity[5]);
+
+        // truncate[1]: X'00FF' and X'0001' share the partition X'00', 'a' and 'ab' the
+        // partition 'a'.
+        auto truncated =
+                channels("truncate[1]", method,
+                         {std::string("\x00\xff", 2), std::string("\x00\x01", 2), "a", "ab"});
+        ASSERT_EQ(4, truncated.size());
+        EXPECT_EQ(truncated[0], truncated[1]);
+        EXPECT_EQ(truncated[2], truncated[3]);
+    }
+}
+
 TEST_F(ExternalTableSinkHashPartitionerTest, PaimonFixedBucketRejectsMissingMetadata) {
     TExternalTableSinkHashPartitionInfo info;
     info.__set_partition_function("paimon_fixed_bucket");

fe/fe-common/src/main/java/org/apache/doris/common/Config.java

@@ -2030,9 +2030,11 @@ public class Config extends ConfigBase {
     public static final int MAP_AGG_V2_EXACT_FRAME_MIN_BE_EXEC_VERSION = 17;
     // Older backends reject the iceberg_transform external sink partition function.
     public static final int ICEBERG_TRANSFORM_ROUTING_MIN_BE_EXEC_VERSION = 18;
+    // Older backends cannot hash VARBINARY values in a shuffle or partition exchange.
+    public static final int VARBINARY_SHUFFLE_HASH_MIN_BE_EXEC_VERSION = 19;
 
     @ConfField(mutable = false)
-    public static int max_be_exec_version = ICEBERG_TRANSFORM_ROUTING_MIN_BE_EXEC_VERSION;
+    public static int max_be_exec_version = VARBINARY_SHUFFLE_HASH_MIN_BE_EXEC_VERSION;
 
     /**
      * Min data version of backends serialize block.