branch-4.1: [fix](paimon) Support Variant in native reader

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;
 
@@ -235,6 +250,8 @@ class ColumnVariantV2 final : public COWHelper<IColumn, ColumnVariantV2> {
     ColumnVariantV2(const ColumnVariantV2& other);
 
     uint32_t _find_or_insert_metadata(StringRef metadata);
+    void _replace_shredded_state_with(const ColumnVariantV2& replacement);
+    void _ensure_serialized();
     void _adopt_state_from(ColumnVariantV2& replacement);
     void _detach_metadata_for_write();
     void _check_invariants() 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.

be/src/format_v2/file_reader.cpp

@@ -74,7 +74,10 @@ std::string FileScanRequest::debug_string() const {
         out << column_id << ":" << block_position;
     }
     out << "}, conjunct_count=" << conjuncts.size()
-        << ", delete_conjunct_count=" << delete_conjuncts.size()
+        << ", delete_conjunct_count=" << delete_conjuncts.size() << ", variant_schema_overrides="
+        << join_debug_strings(
+                   variant_schema_overrides,
+                   [](const LocalColumnIndex& projection) { return projection.debug_string(); })
         << ", count_star_placeholder_columns={";
     const char* delimiter = "";
     for (const auto column_id : count_star_placeholder_columns) {

be/src/format_v2/file_reader.h

@@ -94,6 +94,11 @@ struct FileScanRequest {
     // predicate_columns, the value is semantically required and must still be validated and read.
     std::vector<LocalColumnId> count_star_placeholder_columns;
 
+    // Table formats may assign semantics that legacy physical files do not encode. Each path here
+    // identifies an unannotated Parquet group that the physical reader must validate and decode as
+    // Variant. Keeping this explicit prevents generic Parquet scans from guessing based on names.
+    std::vector<LocalColumnIndex> variant_schema_overrides;
+
     bool is_count_star_placeholder(LocalColumnId column_id) const {
         return std::ranges::find(count_star_placeholder_columns, column_id) !=
                count_star_placeholder_columns.end();

be/src/format_v2/parquet/native_schema_desc.cpp

@@ -65,6 +65,8 @@ enum class VariantPrimitiveAnnotation : uint8_t {
     NONE,
     INT8,
     INT16,
+    INT32,
+    INT64,
     DECIMAL,
     DATE,
     TIME_MICROS,
@@ -91,6 +93,15 @@ static VariantPrimitiveAnnotation variant_logical_annotation(
         if (logical.INTEGER.bitWidth == 16) {
             return VariantPrimitiveAnnotation::INT16;
         }
+        // Iceberg 1.11 writes full-width signed INTEGER annotations even though the Variant
+        // specification represents these widths without an annotation. Keep the widths distinct
+        // so validation only accepts them with the matching physical type.
+        if (logical.INTEGER.bitWidth == 32) {
+            return VariantPrimitiveAnnotation::INT32;
+        }
+        if (logical.INTEGER.bitWidth == 64) {
+            return VariantPrimitiveAnnotation::INT64;
+        }
         return VariantPrimitiveAnnotation::UNSUPPORTED;
     }
     if (logical.__isset.DECIMAL) {
@@ -136,6 +147,11 @@ static VariantPrimitiveAnnotation variant_converted_annotation(
         return VariantPrimitiveAnnotation::INT8;
     case tparquet::ConvertedType::INT_16:
         return VariantPrimitiveAnnotation::INT16;
+    // Parquet Java mirrors full-width logical annotations into these legacy converted types.
+    case tparquet::ConvertedType::INT_32:
+        return VariantPrimitiveAnnotation::INT32;
+    case tparquet::ConvertedType::INT_64:
+        return VariantPrimitiveAnnotation::INT64;
     case tparquet::ConvertedType::DECIMAL:
         return VariantPrimitiveAnnotation::DECIMAL;
     case tparquet::ConvertedType::DATE:
@@ -215,11 +231,13 @@ static Status validate_variant_primitive_type(const NativeFieldSchema& typed) {
         valid = annotation == VariantPrimitiveAnnotation::NONE ||
                 annotation == VariantPrimitiveAnnotation::INT8 ||
                 annotation == VariantPrimitiveAnnotation::INT16 ||
+                annotation == VariantPrimitiveAnnotation::INT32 ||
                 annotation == VariantPrimitiveAnnotation::DECIMAL ||
                 annotation == VariantPrimitiveAnnotation::DATE;
         break;
     case tparquet::Type::INT64:
         valid = annotation == VariantPrimitiveAnnotation::NONE ||
+                annotation == VariantPrimitiveAnnotation::INT64 ||
                 annotation == VariantPrimitiveAnnotation::DECIMAL ||
                 annotation == VariantPrimitiveAnnotation::TIME_MICROS ||
                 annotation == VariantPrimitiveAnnotation::TIMESTAMP_MICROS ||
@@ -266,17 +284,22 @@ class ScopedBoolOverride {
     bool _original;
 };
 
-static Status validate_variant_layout(const tparquet::SchemaElement& group_schema,
-                                      const NativeFieldSchema& group_field) {
-    const auto& annotation = group_schema.logicalType.VARIANT;
-    if (annotation.__isset.specification_version && annotation.specification_version != 1) {
+Status validate_variant_layout(const NativeFieldSchema& group_field,
+                               std::optional<int8_t> specification_version,
+                               bool allow_optional_shredded_metadata) {
+    if (specification_version.has_value() && *specification_version != 1) {
         return Status::NotSupported("Parquet Variant specification version {} is not supported",
-                                    annotation.specification_version);
+                                    *specification_version);
+    }
+    if (group_field.parquet_schema.__isset.repetition_type &&
+        group_field.parquet_schema.repetition_type == tparquet::FieldRepetitionType::REPEATED) {
+        return Status::NotSupported("repeated Parquet Variant group {} is not supported",
+                                    group_field.name);
     }
     if (group_field.children.size() < 2 || group_field.children.size() > 3) {
         return Status::Corruption(
                 "Parquet Variant {} must contain metadata, value, and optional typed_value",
-                group_schema.name);
+                group_field.name);
     }
 
     const NativeFieldSchema* metadata = nullptr;
@@ -292,22 +315,29 @@ static Status validate_variant_layout(const tparquet::SchemaElement& group_schem
             target = &typed_value;
         } else {
             return Status::Corruption("Parquet Variant {} has unexpected child {}",
-                                      group_schema.name, child.name);
+                                      group_field.name, child.name);
         }
         if (*target != nullptr) {
-            return Status::Corruption("Parquet Variant {} has duplicate child {}",
-                                      group_schema.name, child.name);
+            return Status::Corruption("Parquet Variant {} has duplicate child {}", group_field.name,
+                                      child.name);
         }
         *target = &child;
     }
     if (metadata == nullptr || value == nullptr) {
         return Status::Corruption("Parquet Variant {} requires metadata and value children",
-                                  group_schema.name);
-    }
+                                  group_field.name);
+    }
+    const auto metadata_repetition = metadata->parquet_schema.repetition_type;
+    // Paimon makes every field in its shredded carrier optional. Restrict that compatibility to
+    // unannotated overrides; row materialization still rejects null metadata for a non-null value.
+    const bool valid_metadata_repetition =
+            metadata_repetition == tparquet::FieldRepetitionType::REQUIRED ||
+            (allow_optional_shredded_metadata && typed_value != nullptr &&
+             metadata_repetition == tparquet::FieldRepetitionType::OPTIONAL);
     if (!metadata->children.empty() || metadata->physical_type != tparquet::Type::BYTE_ARRAY ||
-        metadata->parquet_schema.repetition_type != tparquet::FieldRepetitionType::REQUIRED) {
+        !valid_metadata_repetition) {
         return Status::Corruption("Parquet Variant {} metadata must be a required BYTE_ARRAY",
-                                  group_schema.name);
+                                  group_field.name);
     }
     const auto expected_value_repetition = typed_value == nullptr
                                                    ? tparquet::FieldRepetitionType::REQUIRED
@@ -317,13 +347,13 @@ static Status validate_variant_layout(const tparquet::SchemaElement& group_schem
     if (!value->children.empty() || value->physical_type != tparquet::Type::BYTE_ARRAY ||
         value->parquet_schema.repetition_type != expected_value_repetition) {
         return Status::Corruption("Parquet Variant {} value must be a {} BYTE_ARRAY",
-                                  group_schema.name,
+                                  group_field.name,
                                   typed_value == nullptr ? "required" : "optional");
     }
     if (typed_value != nullptr &&
         typed_value->parquet_schema.repetition_type != tparquet::FieldRepetitionType::OPTIONAL) {
         return Status::Corruption("Parquet Variant {} typed_value must be optional",
-                                  group_schema.name);
+                                  group_field.name);
     }
 
     enum class WrapperContext : uint8_t { OBJECT_FIELD, ARRAY_ELEMENT };
@@ -945,7 +975,12 @@ Status NativeFieldDescriptor::parse_group_field(
             ScopedBoolOverride timestamp_tz_mapping(_enable_mapping_timestamp_tz, true);
             RETURN_IF_ERROR(parse_struct_field(t_schemas, curr_pos, group_field));
         }
-        RETURN_IF_ERROR(validate_variant_layout(group_schema, *group_field));
+        const auto& annotation = group_schema.logicalType.VARIANT;
+        const auto specification_version =
+                annotation.__isset.specification_version
+                        ? std::optional<int8_t>(annotation.specification_version)
+                        : std::nullopt;
+        RETURN_IF_ERROR(validate_variant_layout(*group_field, specification_version));
         group_field->variant_physical_type = group_field->data_type;
         // Native page readers dispatch groups from data_type, so preserve the physical STRUCT
         // here. The public Parquet schema maps it to logical Variant without losing this shape.

be/src/format_v2/parquet/native_schema_desc.h

@@ -22,6 +22,7 @@
 #include <stddef.h>
 #include <stdint.h>
 
+#include <optional>
 #include <string>
 #include <unordered_map>
 #include <unordered_set>
@@ -86,6 +87,10 @@ struct NativeFieldSchema {
     uint64_t get_max_column_id() const;
 };
 
+Status validate_variant_layout(const NativeFieldSchema& group_field,
+                               std::optional<int8_t> specification_version = std::nullopt,
+                               bool allow_optional_shredded_metadata = false);
+
 // V2 owns this schema tree and parser so footer/schema planning never invokes the V1 reader path.
 class NativeFieldDescriptor {
 private: