[fix](variant) Preserve projected shredded states across exchange

be/src/core/column/variant_v2/column_variant_v2.h

@@ -53,6 +53,9 @@ struct VariantShreddedTypedValue {
     // retain the decoded leaf without copying it or depending on scanner lifetime.
     ColumnPtr column;
     DataTypePtr type;
+    // Physical identities such as binary annotations cannot use the typed scalar state. In that
+    // case the format reader may return an exact Nullable<VariantV2> leaf instead.
+    ColumnPtr normalized;
 };
 
 // Format readers keep their native shredded representation behind this interface. Core Variant
@@ -76,15 +79,26 @@ class VariantShreddedState {
                                                                size_t length) const = 0;
     virtual std::shared_ptr<VariantShreddedState> select_indices(
             const uint32_t* indices_begin, const uint32_t* indices_end) const = 0;
+    // False means the state contains only projected leaves and cannot reconstruct root values.
+    virtual bool can_materialize() const = 0;
     // Appends another state only when both format-owned physical layouts have identical semantics.
     // An incompatible source must leave this state unchanged and return false.
     virtual bool try_append(const VariantShreddedState& source) = 0;
     virtual std::optional<VariantShreddedTypedValue> find_typed_value(
             std::span<const VariantShreddedPathSegment> path) const = 0;
+    // Produces an exact Variant representation of one requested path. Complete states may fall
+    // back to their canonical roots; projected states must preserve the format's physical scalar
+    // identity rather than inferring it from the decoded value.
+    virtual std::optional<ColumnPtr> find_normalized_value(
+            std::span<const VariantShreddedPathSegment> path) const = 0;
 
     // The returned column is cached and owned by this state, so borrowed VariantRef values remain
     // valid for the state lifetime. Implementations must not materialize before this is called.
     virtual const ColumnVariantV2& materialized_column() const = 0;
+    // Whole-column transport may encode only the retained projection because access-path planning
+    // guarantees that omitted fields have no downstream consumer. The returned column must be a
+    // self-contained, non-shredded wire representation with the same row count.
+    virtual const ColumnVariantV2& serialized_column() const = 0;
 };
 
 // ColumnVariantV2 stores a whole column in exactly one state: encoded Variant bytes, one nullable
@@ -146,6 +160,7 @@ class ColumnVariantV2 final : public COWHelper<IColumn, ColumnVariantV2> {
     const DataTypePtr& typed_type() const;
     std::optional<VariantShreddedTypedValue> find_shredded_typed_value(
             std::span<const VariantShreddedPathSegment> path) const;
+    const ColumnVariantV2& serialization_column() const;
     void ensure_encoded();
     ReadView read_view() const;
 

be/src/core/data_type_serde/data_type_variant_v2_serde.cpp

@@ -181,7 +181,7 @@ DataTypeVariantV2SerDe::DataTypeVariantV2SerDe(int nesting_level) : DataTypeSerD
 
 int64_t DataTypeVariantV2SerDe::get_uncompressed_serialized_bytes(const IColumn& column,
                                                                   int be_exec_version) {
-    const auto& variant = get_variant_v2_column(column);
+    const auto& variant = get_variant_v2_column(column).serialization_column();
     int64_t size = sizeof(bool) + sizeof(size_t) * 2 + sizeof(bool);
     if (variant.is_typed()) {
         const DataTypePtr nullable_type = make_nullable(variant._typed_type);
@@ -199,7 +199,7 @@ char* DataTypeVariantV2SerDe::serialize(const IColumn& column, char* buf, int be
     const IColumn* physical = &column;
     size_t saved_rows = 0;
     buf = serialize_const_flag_and_row_num(&physical, buf, &saved_rows);
-    const auto& variant = assert_cast<const ColumnVariantV2&>(*physical);
+    const auto& variant = assert_cast<const ColumnVariantV2&>(*physical).serialization_column();
     DCHECK_EQ(variant.size(), saved_rows);
     unaligned_store<bool>(buf, variant.is_typed());
     buf += sizeof(bool);

be/src/core/value/variant/variant_batch_builder.cpp

@@ -602,10 +602,18 @@ class VariantCollectionCore {
             add_bool(value.get_bool());
             return;
         case VariantPrimitiveId::INT8:
+            add_scalar(VariantScalarRef::integer(value.get_int(), 1));
+            return;
         case VariantPrimitiveId::INT16:
+            add_scalar(VariantScalarRef::integer(value.get_int(), 2));
+            return;
         case VariantPrimitiveId::INT32:
+            add_scalar(VariantScalarRef::integer(value.get_int(), 4));
+            return;
         case VariantPrimitiveId::INT64:
-            add_int(value.get_int());
+            // Importing an existing VariantRef must retain its physical identity; external
+            // shredded schemas use the width to distinguish otherwise equal scalar values.
+            add_scalar(VariantScalarRef::integer(value.get_int(), 8));
             return;
         case VariantPrimitiveId::FLOAT:
             add_scalar(VariantScalarRef::float32(value.get_float()));
@@ -621,7 +629,7 @@ class VariantCollectionCore {
                 throw Exception(ErrorCode::CORRUPTION,
                                 "Variant imported decimal exceeds precision 38");
             }
-            add_scalar(VariantScalarRef::decimal(decimal.unscaled, decimal.scale));
+            add_scalar(VariantScalarRef::decimal(decimal.unscaled, decimal.scale, decimal.width));
             return;
         }
         case VariantPrimitiveId::DATE:

be/src/exprs/function/function_variant_element_v2.cpp

@@ -152,14 +152,19 @@ std::optional<ColumnPtr> extract_shredded_typed_variant_element(
     if (!match.has_value()) {
         return std::nullopt;
     }
-    const auto& leaf = assert_cast<const ColumnNullable&>(*match->column);
+    const ColumnPtr& matched_column = match->normalized ? match->normalized : match->column;
+    const auto& leaf = assert_cast<const ColumnNullable&>(*matched_column);
     auto nulls = leaf.get_null_map_column().clone_resized(source.size());
     auto& null_data = assert_cast<ColumnUInt8&>(*nulls).get_data();
     for (size_t row = 0; row < source.size(); ++row) {
         null_data[row] =
                 static_cast<uint8_t>(null_data[row] != 0 || is_outer_null(outer_nulls, row));
     }
 
+    if (match->normalized) {
+        return ColumnNullable::create(leaf.get_nested_column_ptr(), std::move(nulls));
+    }
+
     // The typed ColumnVariantV2 retains the exact decoded Parquet leaf. Only the SQL result null
     // map is produced here, so predicates and casts can consume the leaf without reconstructing
     // canonical Variant rows.