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" }