branch-4.1:[opt](variant) Seek unshredded Parquet Variant paths directly

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

@@ -527,12 +527,66 @@ class CompositeVariantShreddedState final : public VariantShreddedState {
                                               .normalized = nullptr};
         }
 
-        auto normalized = find_normalized_value(path);
-        if (!normalized.has_value()) {
-            return std::nullopt;
+        const bool all_direct_values =
+                all_direct && std::ranges::all_of(matches, [](const auto& match) {
+                    const bool typed = match.column && match.type && !match.normalized;
+                    const bool normalized = match.normalized && !match.column && !match.type;
+                    return typed || normalized;
+                });
+        if (all_direct_values) {
+            auto values = ColumnVariantV2::create();
+            auto nulls = ColumnUInt8::create();
+            nulls->reserve(size());
+            for (size_t index = 0; index < matches.size(); ++index) {
+                const auto& match = matches[index];
+                std::optional<ColumnPtr> normalized_typed;
+                if (!match.normalized) {
+                    // Typed Parquet leaves need their segment's schema-aware normalization to
+                    // preserve physical distinctions such as INT32 versus INT64. Normalize only
+                    // those segments; a normalized unshredded match already contains exact bytes.
+                    normalized_typed = _segments[index]->find_normalized_value(path);
+                    if (!normalized_typed.has_value()) {
+                        return std::nullopt;
+                    }
+                }
+                const ColumnPtr& matched = match.normalized ? match.normalized : *normalized_typed;
+                const auto& nullable = assert_cast<const ColumnNullable&>(*matched);
+                nulls->insert_range_from(nullable.get_null_map_column(), 0, nullable.size());
+                const auto& variants =
+                        assert_cast<const ColumnVariantV2&>(nullable.get_nested_column());
+                values->insert_range_from(variants, 0, variants.size());
+            }
+            return VariantShreddedTypedValue {
+                    .column = nullptr,
+                    .type = nullptr,
+                    .normalized = ColumnNullable::create(std::move(values), std::move(nulls))};
+        }
+
+        auto values = ColumnVariantV2::create();
+        auto nulls = ColumnUInt8::create();
+        nulls->reserve(size());
+        for (size_t index = 0; index < _segments.size(); ++index) {
+            std::optional<ColumnPtr> normalized;
+            if (index < matches.size() && matches[index].normalized) {
+                // Keep normalized prefix matches that were completed before a later segment
+                // requested fallback. In particular, do not seek an unshredded segment twice.
+                normalized = matches[index].normalized;
+            } else {
+                normalized = _segments[index]->find_normalized_value(path);
+            }
+            if (!normalized.has_value()) {
+                return std::nullopt;
+            }
+            const auto& nullable = assert_cast<const ColumnNullable&>(**normalized);
+            const auto& variants =
+                    assert_cast<const ColumnVariantV2&>(nullable.get_nested_column());
+            values->insert_range_from(variants, 0, variants.size());
+            nulls->insert_range_from(nullable.get_null_map_column(), 0, nullable.size());
         }
         return VariantShreddedTypedValue {
-                .column = nullptr, .type = nullptr, .normalized = std::move(*normalized)};
+                .column = nullptr,
+                .type = nullptr,
+                .normalized = ColumnNullable::create(std::move(values), std::move(nulls))};
     }
 
     std::optional<ColumnPtr> find_normalized_value(
@@ -1023,8 +1077,10 @@ void ColumnVariantV2::EncodedRowsAppender::append( // NOLINT(readability-functio
             validate_variant_metadata(value.metadata);
             unique_metadatas.push_back(value.metadata);
         } else if (unique_metadatas.size() == 1 && metadata_ids_by_value.empty() &&
-                   StringRef(unique_metadatas.front().data, unique_metadatas.front().size) ==
-                           StringRef(value.metadata.data, value.metadata.size)) {
+                   unique_metadatas.front().size == value.metadata.size &&
+                   (unique_metadatas.front().data == value.metadata.data ||
+                    StringRef(unique_metadatas.front().data, unique_metadatas.front().size) ==
+                            StringRef(value.metadata.data, value.metadata.size))) {
             // Iceberg files normally share one metadata dictionary across a batch. Avoid a hash
             // table and per-row ids until a second distinct dictionary is actually observed.
         } else {

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

@@ -131,6 +131,40 @@ void require_valid_primitive(VariantRef value) {
 
 void require_exact_value(VariantRef value, uint32_t depth);
 
+VariantBasicType require_shallow_value(VariantRef value, uint32_t depth) {
+    if (depth > VARIANT_MAX_NESTING_DEPTH) {
+        throw Exception(ErrorCode::CORRUPTION,
+                        "VariantField value exceeds maximum nesting depth {}",
+                        VARIANT_MAX_NESTING_DEPTH);
+    }
+    const size_t encoded_size = value.value_size();
+    if (encoded_size != value.value.size) {
+        throw Exception(ErrorCode::CORRUPTION,
+                        "VariantField value has {} trailing bytes after its {} byte root",
+                        value.value.size - encoded_size, encoded_size);
+    }
+
+    const VariantBasicType type = value.basic_type();
+    if ((type == VariantBasicType::OBJECT || type == VariantBasicType::ARRAY) &&
+        depth == VARIANT_MAX_NESTING_DEPTH && value.num_elements() != 0) {
+        throw Exception(ErrorCode::CORRUPTION,
+                        "VariantField value exceeds maximum nesting depth {}",
+                        VARIANT_MAX_NESTING_DEPTH);
+    }
+    switch (type) {
+    case VariantBasicType::PRIMITIVE:
+        require_valid_primitive(value);
+        break;
+    case VariantBasicType::SHORT_STRING:
+        require_valid_utf8(value.get_string(), "short string");
+        break;
+    case VariantBasicType::OBJECT:
+    case VariantBasicType::ARRAY:
+        break;
+    }
+    return type;
+}
+
 struct ObjectValueSpan {
     size_t offset;
     size_t size;
@@ -201,24 +235,9 @@ void require_valid_array(VariantRef value, uint32_t depth) {
 }
 
 void require_exact_value(VariantRef value, uint32_t depth) {
-    if (depth > VARIANT_MAX_NESTING_DEPTH) {
-        throw Exception(ErrorCode::CORRUPTION,
-                        "VariantField value exceeds maximum nesting depth {}",
-                        VARIANT_MAX_NESTING_DEPTH);
-    }
-    const size_t encoded_size = value.value_size();
-    if (encoded_size != value.value.size) {
-        throw Exception(ErrorCode::CORRUPTION,
-                        "VariantField value has {} trailing bytes after its {} byte root",
-                        value.value.size - encoded_size, encoded_size);
-    }
-
-    switch (value.basic_type()) {
+    switch (require_shallow_value(value, depth)) {
     case VariantBasicType::PRIMITIVE:
-        require_valid_primitive(value);
-        return;
     case VariantBasicType::SHORT_STRING:
-        require_valid_utf8(value.get_string(), "short string");
         return;
     case VariantBasicType::OBJECT:
         require_valid_object(value, depth);
@@ -278,8 +297,12 @@ void validate_variant_metadata(VariantMetadataRef metadata) {
 }
 
 void validate_variant_payload(VariantRef value) {
+    validate_variant_payload(value, 0);
+}
+
+void validate_variant_payload(VariantRef value, uint32_t initial_depth) {
     require_non_null(value.value, "value");
-    require_exact_value(value, 0);
+    require_exact_value(value, initial_depth);
 }
 
 VariantField::VariantField(std::unique_ptr<char[]> data, size_t size) noexcept

be/src/core/value/variant/variant_field.h

@@ -35,9 +35,12 @@ using VariantMap = std::map<PathInData, FieldWithDataType>;
 // validating one or more payloads that reference the same dictionary.
 void validate_variant_metadata(VariantMetadataRef metadata);
 
-// Validate exactly one recursive Variant payload. The referenced metadata must already have
-// passed validate_variant_metadata().
+// Validate exactly one recursive Variant payload. The referenced metadata must conform to the
+// Parquet Variant specification; callers accepting standalone untrusted rows should validate the
+// complete metadata dictionary first. The overload preserves a subtree's original depth when it
+// was selected from a larger Variant value.
 void validate_variant_payload(VariantRef value);
+void validate_variant_payload(VariantRef value, uint32_t initial_depth);
 
 // Holds the legacy V1 path map or owns one encoded V2 row. The encoded byte layout is
 // [u32 little-endian metadata_size][metadata][exactly one value].