From d62762ecee782903eae16050f1d2989afe81a1c9 Mon Sep 17 00:00:00 2001 From: morningman Date: Thu, 15 Oct 2026 14:36:05 +0100 Subject: [PATCH] [fix](be) Hash VARBINARY values in shuffle and partition exchanges ### What problem does this PR solve? Issue Number: None Related PR: #68921 Problem Summary: ColumnVarbinary inherited IColumn's unsupported CRC32 and CRC32C hash methods, so every exchange that partitions rows by a VARBINARY value failed. Since #68786 maps Iceberg BINARY and FIXED to VARBINARY: - MERGE INTO an Iceberg table partitioned by identity(binary_col) or truncate(n, binary_col) fails with "update_crc32c_batch is not supported for ColumnVarbinary" once its source has several instances. - #68921 routes a partitioned Iceberg INSERT by the transformed partition value, but keeps random writes for a BINARY or FIXED value for the same reason, so every writer writes a file for every such partition it receives. - A shuffle join on ARRAY keys fails with "update_crc32c_single is not supported for ColumnVarbinary", while the same join as a broadcast join works. ColumnVarbinary now implements update_crcs_with_value, update_crc_with_value, update_crc32c_batch and update_crc32c_single by hashing the value's bytes as ColumnString does, so the same bytes get the same hash in both columns. SipHash and xxHash stay unsupported. BE exec version 19 marks the capability, and the Iceberg connector routes a BINARY or FIXED partition value only when the statement's exec version is at least 19; an older backend keeps random writes. ### Release note MERGE INTO an Iceberg table with an identity or truncate partition on a binary column no longer fails, and INSERT into such a table routes the rows of a partition to one writer. ### Check List (For Author) - Test: Unit Test, Regression test - BE UT: ColumnVarbinaryTest (CRC methods match ColumnString, ARRAY), ExternalTableSinkHashPartitionerTest (Iceberg identity / truncate routing of VARBINARY values) - FE UT: IcebergWritePlanProviderTest - Regression: test_iceberg_write_binary_partitions, test_varbinary_sql_support - Behavior changed: Yes. MERGE into binary partitions works, INSERT routes binary partition values on backends at exec version 19, and a shuffle join on ARRAY keys works. - Does this need documentation: No Co-Authored-By: Claude Opus 5.5 (1M context) --- be/src/agent/be_exec_version_manager.cpp | 4 +- be/src/agent/be_exec_version_manager.h | 1 + be/src/core/column/column_varbinary.cpp | 70 +++++++++++ be/src/core/column/column_varbinary.h | 18 ++- be/test/core/column/column_varbinary_test.cpp | 110 ++++++++++++++++-- ...ernal_table_sink_hash_partitioner_test.cpp | 72 +++++++++++- .../java/org/apache/doris/common/Config.java | 4 +- .../iceberg/IcebergWritePlanProvider.java | 18 +-- .../iceberg/IcebergWritePlanProviderTest.java | 50 +++++--- .../test_varbinary_sql_support.out | 28 +++++ .../test_iceberg_write_binary_partitions.out | 32 +++++ .../test_varbinary_sql_support.groovy | 9 ++ ...est_iceberg_write_binary_partitions.groovy | 37 +++++- 13 files changed, 415 insertions(+), 38 deletions(-) diff --git a/be/src/agent/be_exec_version_manager.cpp b/be/src/agent/be_exec_version_manager.cpp index 960159e8d15e61..c546960b25d56c 100644 --- a/be/src/agent/be_exec_version_manager.cpp +++ b/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> BeExecVersionManager::_function_change_map {}; std::map> BeExecVersionManager::_function_alternative_map {}; diff --git a/be/src/agent/be_exec_version_manager.h b/be/src/agent/be_exec_version_manager.h index 53564005f0ecf8..b40933effddc03 100644 --- a/be/src/agent/be_exec_version_manager.h +++ b/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: diff --git a/be/src/core/column/column_varbinary.cpp b/be/src/core/column/column_varbinary.cpp index 71dd2f91e6b07f..d393a6f3d91b9a 100644 --- a/be/src/core/column/column_varbinary.cpp +++ b/be/src/core/column/column_varbinary.cpp @@ -17,6 +17,7 @@ #include "core/column/column_varbinary.h" +#include #include #include @@ -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(_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(_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(_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(_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(keys[i].data + keys[i].size), i); diff --git a/be/src/core/column/column_varbinary.h b/be/src/core/column/column_varbinary.h index 2efc8da06fd048..455c04eb632011 100644 --- a/be/src/core/column/column_varbinary.h +++ b/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 +// 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 { private: using Self = ColumnVarbinary; @@ -123,6 +126,19 @@ class ColumnVarbinary final : public COWHelper { 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; diff --git a/be/test/core/column/column_varbinary_test.cpp b/be/test/core/column/column_varbinary_test.cpp index 20e07dfa8af27f..1aae5ec57ad19f 100644 --- a/be/test/core/column/column_varbinary_test.cpp +++ b/be/test/core/column/column_varbinary_test.cpp @@ -26,12 +26,16 @@ #include #include #include +#include #include #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 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 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(values.size()); + // Distinct seeds, so a method that ignored the bytes or the row would not match by accident. + auto seeds = [&]() { + std::vector hashes(rows); + for (uint32_t i = 0; i < rows; ++i) { + hashes[i] = 7 * i + 1; + } + return hashes; + }; + + for (const uint8_t* nulls : {static_cast(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> {{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 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 binary_crc32cs(3, 5); + std::vector 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 binary_crcs(3, 5); + std::vector 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) { diff --git a/be/test/exec/partitioner/external_table_sink_hash_partitioner_test.cpp b/be/test/exec/partitioner/external_table_sink_hash_partitioner_test.cpp index c28e9a31eda4f9..fdfb6592c1e063 100644 --- a/be/test/exec/partitioner/external_table_sink_hash_partitioner_test.cpp +++ b/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& 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(), + "key")); + return block; + } + + std::vector channels(const std::string& transform, + ShuffleHashMethod method, + const std::vector& 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 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"); diff --git a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java index 7a2df0152c4441..aea436e5505b8e 100644 --- a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java +++ b/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. diff --git a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWritePlanProvider.java b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWritePlanProvider.java index 5a06993830c678..d929c5938cdd7d 100644 --- a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWritePlanProvider.java +++ b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWritePlanProvider.java @@ -114,6 +114,8 @@ public ConnectorRowChangeStyle getRowChangeStyle() { private static final int SUPPORT_NESTED_PARTITION_WRITE_EXEC_VERSION = 12; // Older backends reject the iceberg_transform external sink partition function. private static final int SUPPORT_TRANSFORM_ROUTING_EXEC_VERSION = 18; + // Older backends cannot hash a BINARY or FIXED value (ColumnVarbinary) in the exchange. + private static final int SUPPORT_BINARY_ROUTING_EXEC_VERSION = 19; private static final String TRANSFORM_PARTITION_FUNCTION = "iceberg_transform"; // Legacy IcebergUtils compression-codec property keys (connector-local copies; iceberg SDK has no @@ -626,8 +628,8 @@ transform, parseTransformParam(transform), sourceColumnName, * {@code table_sink_partition_write_max_partition_nums_per_writer} open partitions. * *

{@code null} keeps the engine's random parallel write: an unpartitioned table, a partition field on a nested - * source (the route columns are top-level sink outputs), a partition value the backend cannot hash (see the - * loop), or a backend older than the routing function.

+ * source (the route columns are top-level sink outputs), a BINARY or FIXED partition value on a backend that + * cannot hash it (see the loop), or a backend older than the routing function.

*/ @Override public ConnectorWriteDistribution getWriteDistribution(ConnectorSession session, @@ -654,12 +656,14 @@ public ConnectorWriteDistribution getWriteDistribution(ConnectorSession session, if (sourceColumnName == null) { return null; } - // The exchange hashes the transformed value, and the BE column of a BINARY or FIXED value - // (ColumnVarbinary) has no shuffle hash: identity(binary_key) or truncate(1, binary_key) would fail the - // INSERT with "update_crc32c_batch is not supported for ColumnVarbinary" as soon as it has two writers - // to route to. bucket(16, binary_key) yields an INT and is routed. + // The exchange hashes the transformed value. A backend older than SUPPORT_BINARY_ROUTING_EXEC_VERSION + // cannot hash the BE column of a BINARY or FIXED value (ColumnVarbinary): identity(binary_key) or + // truncate(1, binary_key) would fail the INSERT with "update_crc32c_batch is not supported for + // ColumnVarbinary" as soon as it has two writers to route to. bucket(16, binary_key) yields an INT and + // is routed by any backend that routes transforms. Type.TypeID transformedType = field.transform().getResultType(schema.findType(field.sourceId())).typeId(); - if (transformedType == Type.TypeID.BINARY || transformedType == Type.TypeID.FIXED) { + if ((transformedType == Type.TypeID.BINARY || transformedType == Type.TypeID.FIXED) + && session.getBeExecVersion() < SUPPORT_BINARY_ROUTING_EXEC_VERSION) { return null; } routeColumns.add(sourceColumnName); diff --git a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWritePlanProviderTest.java b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWritePlanProviderTest.java index b265391bb179e8..621ef24ef7992c 100644 --- a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWritePlanProviderTest.java +++ b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWritePlanProviderTest.java @@ -1582,34 +1582,52 @@ public void getWriteDistributionUsesTheStatementPinnedSchema() { } @Test - public void getWriteDistributionKeepsRandomWritesForBinaryPartitionValues() { - // The exchange hashes the transformed value, and the BE column of a BINARY or FIXED value has no shuffle - // hash: routing identity(binary_key) fails the INSERT once it has two writers. bucket() turns the bytes - // into an INT, so that one is routed. MUTATION: dropping the transformed-type check -> the first four - // tables route -> red. + public void getWriteDistributionRoutesBinaryPartitionValuesOnlyOnBackendsThatHashThem() { + // The exchange hashes the transformed value. A backend below exec version 19 cannot hash the BE column of a + // BINARY or FIXED value, so routing identity(binary_key) there fails the INSERT once it has two writers; + // such a write keeps random writes. bucket() turns the bytes into an INT, so it is routed from version 18. + // MUTATION: dropping the version check -> identity_binary routes at 18 -> red; skipping BINARY and FIXED + // unconditionally -> nothing routes at 19 -> red. InMemoryCatalog catalog = freshCatalog(); Schema schema = new Schema( Types.NestedField.required(1, "id", Types.IntegerType.get()), Types.NestedField.optional(2, "binary_key", Types.BinaryType.get()), Types.NestedField.optional(3, "fixed_key", Types.FixedType.ofLength(4))); - Map unroutable = new LinkedHashMap<>(); - unroutable.put("identity_binary", PartitionSpec.builderFor(schema).identity("binary_key").build()); - unroutable.put("truncate_binary", PartitionSpec.builderFor(schema).truncate("binary_key", 1).build()); - unroutable.put("identity_fixed", PartitionSpec.builderFor(schema).identity("fixed_key").build()); - // One unhashable field is enough: the route hashes every field of the spec together. - unroutable.put("int_and_binary", + Map binaryValued = new LinkedHashMap<>(); + binaryValued.put("identity_binary", PartitionSpec.builderFor(schema).identity("binary_key").build()); + binaryValued.put("truncate_binary", PartitionSpec.builderFor(schema).truncate("binary_key", 1).build()); + binaryValued.put("identity_fixed", PartitionSpec.builderFor(schema).identity("fixed_key").build()); + // One binary-valued field is enough: the route hashes every field of the spec together. + binaryValued.put("int_and_binary", PartitionSpec.builderFor(schema).identity("id").identity("binary_key").build()); - for (Map.Entry entry : unroutable.entrySet()) { - Table table = catalog.createTable(TableIdentifier.of("db1", entry.getKey()), schema, entry.getValue()); + Map> routeColumns = new LinkedHashMap<>(); + routeColumns.put("identity_binary", Collections.singletonList("binary_key")); + routeColumns.put("truncate_binary", Collections.singletonList("binary_key")); + routeColumns.put("identity_fixed", Collections.singletonList("fixed_key")); + routeColumns.put("int_and_binary", Arrays.asList("id", "binary_key")); + Map transforms = new LinkedHashMap<>(); + transforms.put("identity_binary", "identity"); + transforms.put("truncate_binary", "truncate[1]"); + transforms.put("identity_fixed", "identity"); + transforms.put("int_and_binary", "identity,identity"); + for (Map.Entry entry : binaryValued.entrySet()) { + String name = entry.getKey(); + Table table = catalog.createTable(TableIdentifier.of("db1", name), schema, entry.getValue()); + IcebergTableHandle handle = new IcebergTableHandle("db1", name); Assertions.assertNull(providerFor(table, contextWithStorage()).getWriteDistribution( - sessionFor(table, contextWithStorage()), new IcebergTableHandle("db1", entry.getKey())), - entry.getKey()); + sessionFor(table, contextWithStorage()).beExecVersion(18), handle), name); + ConnectorWriteDistribution distribution = providerFor(table, contextWithStorage()).getWriteDistribution( + sessionFor(table, contextWithStorage()).beExecVersion(19), handle); + Assertions.assertEquals(routeColumns.get(name), distribution.getRouteColumns(), name); + Assertions.assertEquals(Collections.singletonMap("transforms", transforms.get(name)), + distribution.getPartitionFunctionOptions(), name); } Table bucketed = catalog.createTable(TableIdentifier.of("db1", "bucket_binary"), schema, PartitionSpec.builderFor(schema).bucket("binary_key", 16).build()); ConnectorWriteDistribution distribution = providerFor(bucketed, contextWithStorage()).getWriteDistribution( - sessionFor(bucketed, contextWithStorage()), new IcebergTableHandle("db1", "bucket_binary")); + sessionFor(bucketed, contextWithStorage()).beExecVersion(18), + new IcebergTableHandle("db1", "bucket_binary")); Assertions.assertEquals(Collections.singletonList("binary_key"), distribution.getRouteColumns()); Assertions.assertEquals(Collections.singletonMap("transforms", "bucket[16]"), distribution.getPartitionFunctionOptions()); diff --git a/regression-test/data/datatype_p0/test_varbinary_sql_support.out b/regression-test/data/datatype_p0/test_varbinary_sql_support.out index a14762b78951fe..6ce2ab246ad152 100644 --- a/regression-test/data/datatype_p0/test_varbinary_sql_support.out +++ b/regression-test/data/datatype_p0/test_varbinary_sql_support.out @@ -50,6 +50,34 @@ payload varbinary(2147483647) Yes false \N 8 AB AB AB 9 FF FF FF +-- !array_key_join_broadcast -- +0 0 +1 1 +2 2 +3 3 +4 4 +5 5 +5 8 +6 6 +7 7 +8 5 +8 8 +9 9 + +-- !array_key_join_shuffle -- +0 0 +1 1 +2 2 +3 3 +4 4 +5 5 +5 8 +6 6 +7 7 +8 5 +8 8 +9 9 + -- !long_bytes -- 2048 100 2147 2048 diff --git a/regression-test/data/external_table_p0/iceberg/write/test_iceberg_write_binary_partitions.out b/regression-test/data/external_table_p0/iceberg/write/test_iceberg_write_binary_partitions.out index e60f362fd80caf..3f1f26616e0ec7 100644 --- a/regression-test/data/external_table_p0/iceberg/write/test_iceberg_write_binary_partitions.out +++ b/regression-test/data/external_table_p0/iceberg/write/test_iceberg_write_binary_partitions.out @@ -281,21 +281,53 @@ 3132 2500 2 9998 3133 2500 3 9999 +-- !binary_multi_writer_identity_INSERT_INTO_files -- +4 + -- !binary_multi_writer_identity_INSERT_OVERWRITE_TABLE -- 3130 2500 0 9996 3131 2500 1 9997 3132 2500 2 9998 3133 2500 3 9999 +-- !binary_multi_writer_identity_INSERT_OVERWRITE_TABLE_files -- +4 + -- !binary_multi_writer_truncate_INSERT_INTO -- 3130 2500 0 9996 3131 2500 1 9997 3132 2500 2 9998 3133 2500 3 9999 +-- !binary_multi_writer_truncate_INSERT_INTO_files -- +1 + -- !binary_multi_writer_truncate_INSERT_OVERWRITE_TABLE -- 3130 2500 0 9996 3131 2500 1 9997 3132 2500 2 9998 3133 2500 3 9999 +-- !binary_multi_writer_truncate_INSERT_OVERWRITE_TABLE_files -- +1 + +-- !binary_merge_identity -- +3130 inserted 2475 100 9996 +3131 inserted 2475 101 9997 +3132 inserted 2475 102 9998 +3133 inserted 2475 103 9999 +3230 updated 25 0 96 +3231 updated 25 1 97 +3232 updated 25 2 98 +3233 updated 25 3 99 + +-- !binary_merge_truncate -- +3130 inserted 2475 100 9996 +3131 inserted 2475 101 9997 +3132 inserted 2475 102 9998 +3133 inserted 2475 103 9999 +3230 updated 25 0 96 +3231 updated 25 1 97 +3232 updated 25 2 98 +3233 updated 25 3 99 + diff --git a/regression-test/suites/datatype_p0/test_varbinary_sql_support.groovy b/regression-test/suites/datatype_p0/test_varbinary_sql_support.groovy index 7715690cd703f8..86463e59e7d59d 100644 --- a/regression-test/suites/datatype_p0/test_varbinary_sql_support.groovy +++ b/regression-test/suites/datatype_p0/test_varbinary_sql_support.groovy @@ -60,6 +60,15 @@ suite("test_varbinary_sql_support") { order_qt_nested_bytes """SELECT id, from_binary(a[1]), from_binary(m['key']), from_binary(s.b) FROM binary_sql_nested""" + // A shuffle join on ARRAY keys hashes the elements' bytes in the exchange, and must match the same + // rows as a broadcast join, which hashes no key in an exchange: ids 5 and 8 hold the same payload. Several + // instances keep the exchange hashing; with one destination it sends every row there without hashing. + sql "set parallel_pipeline_task_num = 4" + for (String distribution : ["broadcast", "shuffle"]) { + "order_qt_array_key_join_${distribution}" """SELECT l.id, r.id FROM binary_sql_nested l + JOIN [${distribution}] binary_sql_nested r ON l.a = r.a""" + } + // Compare every long value while keeping generated output compact across source batches. sql """INSERT INTO binary_sql_input SELECT number + 100, concat(repeat('FF', 700), '00AB') FROM numbers('number'='2048')""" diff --git a/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_binary_partitions.groovy b/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_binary_partitions.groovy index 79995932ee48ca..88e9f47102a84b 100644 --- a/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_binary_partitions.groovy +++ b/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_binary_partitions.groovy @@ -94,9 +94,11 @@ suite("test_iceberg_write_binary_partitions", "p0,external,iceberg,external_dock } // The VALUES writes above have a single-instance source, so their sink has one writer and its exchange - // does not hash. A multi-instance source reaches several writers through a hashing exchange, which must - // not route by a binary partition value: the BE cannot hash it ("update_crc32c_batch is not supported - // for ColumnVarbinary"). Pin the writers, since a fuzzy session can pick one. + // does not hash. A multi-instance source reaches several writers through an exchange that routes each row + // by its binary partition value, hashing the value's bytes, so one writer owns each partition and writes + // its only data file: four files for identity, one for truncate(1), which folds the keys '10' to '13' into + // the partition '1'. Without routing, each of the eight writers writes a file for every partition it + // receives. Pin the writers, since a fuzzy session can pick one. sql "set parallel_pipeline_task_num = 8" for (String transform : ["identity", "truncate"]) { String table = "binary_multi_writer_${transform}" @@ -110,8 +112,37 @@ suite("test_iceberg_write_binary_partitions", "p0,external,iceberg,external_dock SELECT number, CAST(CAST(number % 4 + 10 AS STRING) AS VARBINARY) FROM numbers("number" = "10000")""" "order_qt_${table}_${operation.replace(' ', '_')}" """ SELECT HEX(binary_key), count(*), min(id), max(id) FROM ${table} GROUP BY HEX(binary_key) ORDER BY 1""" + "qt_${table}_${operation.replace(' ', '_')}_files" "SELECT count(*) FROM ${table}\$files" } } + + // MERGE routes the rows it writes by their binary partition value through the same exchange hash, so it + // failed with "update_crc32c_batch is not supported for ColumnVarbinary" before VARBINARY could be hashed. + // Rows 0 to 99 move from the keys '10'..'13' to '20'..'23', and rows 100 to 9999 are inserted. + for (String transform : ["identity", "truncate"]) { + String table = "binary_merge_${transform}" + String expression = transform == "identity" ? "binary_key" : "truncate(1, binary_key)" + sql "DROP TABLE IF EXISTS ${table}" + spark_iceberg """CREATE TABLE demo.${dbName}.${table} (id BIGINT, binary_key BINARY, v STRING) + USING iceberg PARTITIONED BY (${expression}) + TBLPROPERTIES ('format-version'='2', 'write.format.default'='parquet', + 'write.delete.mode'='merge-on-read', 'write.update.mode'='merge-on-read', + 'write.merge.mode'='merge-on-read')""" + sql """INSERT INTO ${table} + SELECT number, CAST(CAST(number % 4 + 10 AS STRING) AS VARBINARY), 'old' FROM numbers("number" = "100")""" + sql """MERGE INTO ${table} t + USING (SELECT number AS id, CAST(CAST(number % 4 + 10 AS STRING) AS VARBINARY) AS binary_key + FROM numbers("number" = "10000")) s + ON t.id = s.id + WHEN MATCHED THEN UPDATE SET binary_key = CAST(CAST(s.id % 4 + 20 AS STRING) AS VARBINARY), v = 'updated' + WHEN NOT MATCHED THEN INSERT (id, binary_key, v) VALUES (s.id, s.binary_key, 'inserted')""" + String summary = "SELECT HEX(binary_key), v, count(*), min(id), max(id) FROM ${table} GROUP BY HEX(binary_key), v" + "order_qt_${table}" summary + spark_iceberg "REFRESH TABLE demo.${dbName}.${table}" + assertSparkDorisResultEquals(spark_iceberg(""" + SELECT hex(binary_key), v, count(*), min(id), max(id) FROM demo.${dbName}.${table} + GROUP BY hex(binary_key), v ORDER BY 1, 2"""), sql("${summary} ORDER BY 1, 2")) + } } finally { sql "SWITCH internal" }