[fix](type) Preserve VARBINARY length in serialized blocks on branch-4.1

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<DataTypeHLL>();
         break;
     case PGenericType::VARBINARY:
-        nested = std::make_shared<DataTypeVarbinary>();
+        nested = std::make_shared<DataTypeVarbinary>(pcolumn.varbinary_length());
         break;
     case PGenericType::LIST:
         DCHECK(pcolumn.children_size() == 1);

be/src/core/data_type/data_type_varbinary.cpp

@@ -17,6 +17,7 @@
 
 #include "core/data_type/data_type_varbinary.h"
 
+#include <gen_cpp/data.pb.h>
 #include <glog/logging.h>
 #include <lz4/lz4.h>
 #include <streamvbyte.h>
@@ -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
 // <size array>: data1 size | data2 size | ... sizeof(size_t) * real_need_copy_num

be/src/core/data_type/data_type_varbinary.h

@@ -81,6 +81,8 @@ class DataTypeVarbinary : public IDataType {
         return std::make_shared<DataTypeVarbinarySerDe>(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);
     }

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<DataTypeVarbinary>(length);
+            if (nullable) {
+                type = make_nullable(type);
+            }
+            auto column = type->create_column();
+            const std::string value = make_bytes(16);
+            column->insert(Field::create_field<TYPE_VARBINARY>(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<const DataTypeVarbinary&>(*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<DataTypeVarbinary>(16);
+    auto binary32 = std::make_shared<DataTypeVarbinary>(32);
+    DataTypeStruct type({std::make_shared<DataTypeArray>(make_nullable(binary16)),
+                         std::make_shared<DataTypeMap>(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<const DataTypeStruct&>(*restored).get_elements();
+    const auto& array = assert_cast<const DataTypeArray&>(*fields[0]);
+    ASSERT_TRUE(array.get_nested_type()->is_nullable());
+    EXPECT_EQ(
+            assert_cast<const DataTypeVarbinary&>(*remove_nullable(array.get_nested_type())).len(),
+            16);
+    const auto& map = assert_cast<const DataTypeMap&>(*fields[1]);
+    EXPECT_EQ(assert_cast<const DataTypeVarbinary&>(*map.get_key_type()).len(), 16);
+    ASSERT_TRUE(map.get_value_type()->is_nullable());
+    EXPECT_EQ(assert_cast<const DataTypeVarbinary&>(*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<const DataTypeVarbinary&>(*restored).len(), -1);
+}
+
 TEST_F(DataTypeVarbinaryTest, ProtobufPreservesDeclaredLength) {
     PTypeDesc type;
     auto* node = type.add_types();

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 {