From b35abbd647e6f81b0cfd68b812bb409194ab54a6 Mon Sep 17 00:00:00 2001 From: Gabriel Date: Sat, 10 Oct 2026 12:15:56 +0800 Subject: [PATCH] [fix](type) Preserve VARBINARY length in serialized blocks --- be/src/core/data_type/data_type_factory.cpp | 2 +- be/src/core/data_type/data_type_varbinary.cpp | 7 ++ be/src/core/data_type/data_type_varbinary.h | 2 + .../data_type/data_type_varbinary_test.cpp | 80 +++++++++++++++++++ gensrc/proto/data.proto | 2 + 5 files changed, 92 insertions(+), 1 deletion(-) diff --git a/be/src/core/data_type/data_type_factory.cpp b/be/src/core/data_type/data_type_factory.cpp index 4ffd9a368a713f..b4bc3380a6429c 100644 --- a/be/src/core/data_type/data_type_factory.cpp +++ b/be/src/core/data_type/data_type_factory.cpp @@ -298,7 +298,7 @@ DataTypePtr DataTypeFactory::create_data_type(const PColumnMeta& pcolumn) { nested = std::make_shared(); break; case PGenericType::VARBINARY: - nested = std::make_shared(); + nested = std::make_shared(pcolumn.varbinary_length()); break; case PGenericType::LIST: DCHECK(pcolumn.children_size() == 1); diff --git a/be/src/core/data_type/data_type_varbinary.cpp b/be/src/core/data_type/data_type_varbinary.cpp index 0bf483b8bbc203..dd7f5c83eef598 100644 --- a/be/src/core/data_type/data_type_varbinary.cpp +++ b/be/src/core/data_type/data_type_varbinary.cpp @@ -17,6 +17,7 @@ #include "core/data_type/data_type_varbinary.h" +#include #include #include #include @@ -54,6 +55,12 @@ bool DataTypeVarbinary::equals(const IDataType& rhs) const { return typeid(rhs) == typeid(*this); } +void DataTypeVarbinary::to_pb_column_meta(PColumnMeta* col_meta) const { + IDataType::to_pb_column_meta(col_meta); + // Iceberg FIXED/UUID writers need the declared width after a Block crosses an exchange. + col_meta->set_varbinary_length(_len); +} + // binary: const flag| row num | real saved num | size array | data array // bool size_t size_t // : data1 size | data2 size | ... sizeof(size_t) * real_need_copy_num diff --git a/be/src/core/data_type/data_type_varbinary.h b/be/src/core/data_type/data_type_varbinary.h index de50e59c8f8c77..acb4afc52b73df 100644 --- a/be/src/core/data_type/data_type_varbinary.h +++ b/be/src/core/data_type/data_type_varbinary.h @@ -81,6 +81,8 @@ class DataTypeVarbinary : public IDataType { return std::make_shared(nesting_level); }; + void to_pb_column_meta(PColumnMeta* col_meta) const override; + void to_protobuf(PTypeDesc* ptype, PTypeNode* node, PScalarType* scalar_type) const override { scalar_type->set_len(_len); } diff --git a/be/test/core/data_type/data_type_varbinary_test.cpp b/be/test/core/data_type/data_type_varbinary_test.cpp index 51da7ed3f349d4..f419c4532de605 100644 --- a/be/test/core/data_type/data_type_varbinary_test.cpp +++ b/be/test/core/data_type/data_type_varbinary_test.cpp @@ -30,13 +30,18 @@ #include "agent/be_exec_version_manager.h" #include "common/exception.h" #include "core/assert_cast.h" +#include "core/block/block.h" #include "core/column/column_const.h" #include "core/column/column_string.h" #include "core/column/column_varbinary.h" #include "core/data_type/common_data_type_serder_test.h" #include "core/data_type/common_data_type_test.h" +#include "core/data_type/data_type_array.h" #include "core/data_type/data_type_factory.hpp" +#include "core/data_type/data_type_map.h" +#include "core/data_type/data_type_nullable.h" #include "core/data_type/data_type_string.h" +#include "core/data_type/data_type_struct.h" #include "core/data_type_serde/data_type_serde.h" #include "core/field.h" #include "core/string_buffer.hpp" @@ -287,6 +292,81 @@ TEST_F(DataTypeVarbinaryTest, ToProtobufDefaultLen) { EXPECT_EQ(scalar.len(), -1); } +TEST_F(DataTypeVarbinaryTest, BlockRoundTripPreservesDeclaredLength) { + for (int length : {-1, 16, 32}) { + for (bool nullable : {false, true}) { + SCOPED_TRACE(::testing::Message() << "length=" << length << ", nullable=" << nullable); + DataTypePtr type = std::make_shared(length); + if (nullable) { + type = make_nullable(type); + } + auto column = type->create_column(); + const std::string value = make_bytes(16); + column->insert(Field::create_field(StringView(value))); + if (nullable) { + column->insert_default(); + } + Block block {{std::move(column), type, "binary_value"}}; + PBlock serialized; + size_t uncompressed_bytes = 0; + size_t compressed_bytes = 0; + int64_t compress_time = 0; + ASSERT_TRUE(block.serialize(BeExecVersionManager::get_newest_version(), &serialized, + &uncompressed_bytes, &compressed_bytes, &compress_time, + segment_v2::CompressionTypePB::SNAPPY) + .ok()); + std::string wire; + ASSERT_TRUE(serialized.SerializeToString(&wire)); + PBlock received; + ASSERT_TRUE(received.ParseFromString(wire)); + Block restored; + int64_t decompress_time = 0; + ASSERT_TRUE(restored.deserialize(received, &uncompressed_bytes, &decompress_time).ok()); + const auto& entry = restored.get_by_position(0); + ASSERT_EQ(entry.type->is_nullable(), nullable); + EXPECT_EQ(assert_cast(*remove_nullable(entry.type)).len(), + length); + ASSERT_EQ(restored.rows(), block.rows()); + for (size_t row = 0; row < block.rows(); ++row) { + EXPECT_EQ((*entry.column)[row], (*block.get_by_position(0).column)[row]); + } + } + } +} + +TEST_F(DataTypeVarbinaryTest, NestedColumnMetadataPreservesDeclaredLength) { + auto binary16 = std::make_shared(16); + auto binary32 = std::make_shared(32); + DataTypeStruct type({std::make_shared(make_nullable(binary16)), + std::make_shared(binary16, make_nullable(binary32))}, + {"array_value", "map_value"}); + PColumnMeta metadata; + type.to_pb_column_meta(&metadata); + std::string wire; + ASSERT_TRUE(metadata.SerializeToString(&wire)); + PColumnMeta received; + ASSERT_TRUE(received.ParseFromString(wire)); + auto restored = DataTypeFactory::instance().create_data_type(received); + const auto& fields = assert_cast(*restored).get_elements(); + const auto& array = assert_cast(*fields[0]); + ASSERT_TRUE(array.get_nested_type()->is_nullable()); + EXPECT_EQ( + assert_cast(*remove_nullable(array.get_nested_type())).len(), + 16); + const auto& map = assert_cast(*fields[1]); + EXPECT_EQ(assert_cast(*map.get_key_type()).len(), 16); + ASSERT_TRUE(map.get_value_type()->is_nullable()); + EXPECT_EQ(assert_cast(*remove_nullable(map.get_value_type())).len(), + 32); +} + +TEST_F(DataTypeVarbinaryTest, LegacyColumnMetadataUsesUnspecifiedLength) { + PColumnMeta metadata; + metadata.set_type(PGenericType::VARBINARY); + auto restored = DataTypeFactory::instance().create_data_type(metadata); + EXPECT_EQ(assert_cast(*restored).len(), -1); +} + TEST_F(DataTypeVarbinaryTest, ProtobufPreservesDeclaredLength) { PTypeDesc type; auto* node = type.add_types(); diff --git a/gensrc/proto/data.proto b/gensrc/proto/data.proto index 6ee532955e417e..66ee8c5e1c10a8 100644 --- a/gensrc/proto/data.proto +++ b/gensrc/proto/data.proto @@ -70,6 +70,8 @@ message PColumnMeta { optional int32 variant_max_subcolumns_count = 10 [default = 0]; optional bool variant_enable_doc_mode = 11 [default = false]; optional bool variant_is_v2 = 12 [default = false]; + // Older blocks omit the declared VARBINARY length and retain the unspecified sentinel. + optional int32 varbinary_length = 13 [default = -1]; } message PBlock {