This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 2d530ac6a76 [fix](be) Hash VARBINARY values in shuffle and partition
exchanges (#69142)
2d530ac6a76 is described below
commit 2d530ac6a76620b9f4bc64d7cac7473b6824a8be
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Fri Oct 16 19:10:53 2026 +0800
[fix](be) Hash VARBINARY values in shuffle and partition exchanges (#69142)
### What problem does this PR solve?
Issue Number: None
Related PR: #68921
Problem Summary:
**In short.** `ColumnVarbinary` has no CRC32 / CRC32C hash, so every
exchange that partitions rows by a VARBINARY value fails. Since #68786
reads Iceberg BINARY and FIXED as VARBINARY, MERGE INTO an Iceberg table
with an `identity` or `truncate` partition on a binary column fails, and
#68921 has to leave INSERT into such a table on random writes. This PR
hashes VARBINARY values by their bytes, exactly as `ColumnString` does,
and lets the Iceberg connector route binary partition values on backends
that can hash them.
**Background.** An exchange that partitions rows hashes the partition
key of each row with CRC32 (`update_crcs_with_value`,
`update_crc_with_value`) or CRC32C (`update_crc32c_batch`,
`update_crc32c_single`): a shuffle join on its keys, an Iceberg INSERT
routed by the transformed partition value (#68921), and MERGE INTO an
Iceberg table, which routes the rows it writes through the same Iceberg
partition function. Array, map and struct keys hash their elements
through the same methods. `ColumnVarbinary` inherits these methods from
`IColumn`, where they throw `NOT_IMPLEMENTED_ERROR`.
**The problem, and what it cost.** On a local single-BE cluster with an
Iceberg REST catalog (tables created by Spark,
`parallel_pipeline_task_num = 8`):
| Statement | master | with this PR |
|---|---|---|
| MERGE INTO a table partitioned by `identity(binary_key)` or
`truncate(1, binary_key)`, with a `numbers()` source | fails:
`update_crc32c_batch is not supported for ColumnVarbinary` | succeeds;
the rows match what Spark reads back |
| INSERT OVERWRITE of 10000 rows into four `identity(binary_key)`
partitions from a `numbers()` source | random writes: 8 data files
(measured with this PR at `be_exec_version` 18, the plan master makes) |
routed: 4 data files, one per partition |
| `SELECT ... JOIN [shuffle] ... ON l.a = r.a` on `ARRAY<VARBINARY>`
keys | fails: `update_crc32c_single is not supported for
ColumnVarbinary` | same rows as `JOIN [broadcast]` |
A scalar VARBINARY comparison is rejected during analysis, so a scalar
VARBINARY never becomes a join key; the shuffle join only reaches the
exchange through a nested type.
**How this PR fixes it.**
- **BE.** `ColumnVarbinary` implements the four CRC methods by hashing
each value's bytes, with and without a null map, as `ColumnString` does,
so the same bytes get the same hash in both columns. SipHash and xxHash
(`update_hash_with_value`, `update_hashes_with_value`,
`update_xxHash_with_value`) stay unsupported, and so does tablet routing
of VARBINARY.
- **Mixed versions.** BE exec version 19 marks the capability and FE
`max_be_exec_version` moves to 19.
`IcebergWritePlanProvider.getWriteDistribution` routes a BINARY or FIXED
transformed partition value only when the statement's exec version is at
least 19; below it the INSERT keeps random writes, as on master today.
`bucket(n, binary_key)` yields an INT and is routed from version 18 as
before. MERGE needs no FE change.
**Results.**
- **BE UT.**
`ColumnVarbinaryTest.ShuffleHashesMatchStringOfTheSameBytes` checks the
four methods against `ColumnString` on empty, NUL, invalid UTF-8 and
inline / out-of-line values, with and without a null map and on a
sub-range; `ArrayOfVarbinaryHashesLikeArrayOfString` does the same
through `ColumnArray` with a NULL element; `ValueHashesAreNotSupported`
keeps SipHash and xxHash unsupported.
`ExternalTableSinkBinaryHashPartitionerTest.IcebergTransformHashesBinaryPartitionValues`
routes `identity` and `truncate[1]` binary values with CRC32 and CRC32C,
so equal partition values share a channel. On master the first two and
the partitioner case fail with `NOT_IMPLEMENTED_ERROR`, since the
methods are inherited from `IColumn`. The three suites (36 cases) pass
under ASAN; clang-tidy reports nothing on the changed files.
- **FE UT.**
`IcebergWritePlanProviderTest#getWriteDistributionRoutesBinaryPartitionValuesOnlyOnBackendsThatHashThem`:
identity / truncate on BINARY, identity on FIXED and a mixed INT +
BINARY spec keep random writes at exec version 18 and are routed at 19;
`bucket(16, binary_key)` is routed at 18. All 1568 tests of
fe-connector-iceberg pass.
- **Regression.** `test_iceberg_write_binary_partitions` now checks the
data files of the multi-writer INSERT and INSERT OVERWRITE (4 for
identity, 1 for truncate(1), which folds the keys into one partition)
and adds MERGE INTO identity / truncate binary partitions, compared with
Spark. `test_varbinary_sql_support` adds a broadcast and a shuffle join
on `ARRAY<VARBINARY>` keys with `parallel_pipeline_task_num = 4`; both
return the same 12 rows, and the shuffle plan hash-partitions both sides
by the array. Both suites pass on a local single-BE cluster against the
Iceberg REST docker environment, where their `.out` files were
generated; only the new blocks were added.
- **Mixed versions.** On the same cluster, with `be_exec_version` set to
18 the INSERT OVERWRITE above writes 8 data files (random writes), and
with 19 it writes 4.
---
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 +++++++++++++++++++--
.../external_table_sink_hash_partitioner_test.cpp | 72 +++++++++++++-
.../main/java/org/apache/doris/common/Config.java | 4 +-
.../iceberg/IcebergWritePlanProvider.java | 18 ++--
.../iceberg/IcebergWritePlanProviderTest.java | 50 +++++++---
.../datatype_p0/test_varbinary_sql_support.out | 28 ++++++
.../write/test_iceberg_write_binary_partitions.out | 32 ++++++
.../datatype_p0/test_varbinary_sql_support.groovy | 9 ++
.../test_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 960159e8d15..c546960b25d 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<std::string, std::set<int>>
BeExecVersionManager::_function_change_map {};
std::map<std::string, std::set<int>>
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 53564005f0e..b40933effdd 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 71dd2f91e6b..d393a6f3d91 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 <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);
diff --git a/be/src/core/column/column_varbinary.h
b/be/src/core/column/column_varbinary.h
index 2efc8da06fd..455c04eb632 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<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 @@ public:
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 20e07dfa8af..1aae5ec57ad 100644
--- a/be/test/core/column/column_varbinary_test.cpp
+++ b/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) {
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 c28e9a31eda..fdfb6592c1e 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 @@ protected:
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");
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 7a2df0152c4..aea436e5505 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 5a06993830c..d929c5938cd 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 class IcebergWritePlanProvider implements
ConnectorWritePlanProvider {
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 @@ public class IcebergWritePlanProvider implements
ConnectorWritePlanProvider {
* {@code table_sink_partition_write_max_partition_nums_per_writer} open
partitions.
*
* <p>{@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.</p>
+ * 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.</p>
*/
@Override
public ConnectorWriteDistribution getWriteDistribution(ConnectorSession
session,
@@ -654,12 +656,14 @@ public class IcebergWritePlanProvider implements
ConnectorWritePlanProvider {
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 b265391bb17..621ef24ef79 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 class IcebergWritePlanProviderTest {
}
@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<String, PartitionSpec> 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<String, PartitionSpec> 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<String, PartitionSpec> entry : unroutable.entrySet()) {
- Table table = catalog.createTable(TableIdentifier.of("db1",
entry.getKey()), schema, entry.getValue());
+ Map<String, List<String>> 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<String, String> 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<String, PartitionSpec> 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 a14762b7895..6ce2ab246ad 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 e60f362fd80..3f1f26616e0 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 7715690cd70..86463e59e7d 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<VARBINARY> 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 79995932ee4..88e9f47102a 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"
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
