Gabriel39 commented on code in PR #66302: URL: https://github.com/apache/doris/pull/66302#discussion_r3689501699
########## be/src/format_v2/parquet/reader/variant_column_reader.cpp: ########## @@ -0,0 +1,1078 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#include "format_v2/parquet/reader/variant_column_reader.h" + +#include <algorithm> +#include <array> +#include <cmath> +#include <cstdint> +#include <cstring> +#include <limits> +#include <mutex> +#include <optional> +#include <string_view> + +#include "common/exception.h" +#include "core/assert_cast.h" +#include "core/column/column_array.h" +#include "core/column/column_decimal.h" +#include "core/column/column_map.h" +#include "core/column/column_nullable.h" +#include "core/column/column_struct.h" +#include "core/column/column_vector.h" +#include "core/column/variant_v2/column_variant_v2.h" +#include "core/column/variant_v2/column_variant_v2_typed_column.h" +#include "core/data_type/data_type_nullable.h" +#include "core/data_type/data_type_variant_v2.h" +#include "core/value/variant/variant_batch_builder.h" +#include "core/value/variant/variant_metadata.h" +#include "format_v2/parquet/parquet_column_schema.h" + +namespace doris::format::parquet { +namespace { + +struct Cell { + const IColumn* column = nullptr; + bool is_null = false; +}; + +Cell cell_at(const IColumn& column, size_t row) { + if (row >= column.size()) { + throw Exception(ErrorCode::CORRUPTION, "Parquet Variant row {} exceeds column size {}", row, + column.size()); + } + if (const auto* nullable = check_and_get_column<ColumnNullable>(column)) { + return {.column = &nullable->get_nested_column(), + .is_null = nullable->get_null_map_data()[row] != 0}; + } + return {.column = &column, .is_null = false}; +} + +const ParquetColumnSchema* find_child(const ParquetColumnSchema& schema, std::string_view name, + size_t* index) { + for (size_t i = 0; i < schema.children.size(); ++i) { + if (schema.children[i]->name == name) { + if (index != nullptr) { + *index = i; + } + return schema.children[i].get(); + } + } + return nullptr; +} + +Cell struct_child_at(const ParquetColumnSchema& schema, const IColumn& physical, size_t row, + std::string_view name, const ParquetColumnSchema** child_schema) { + const auto& structure = assert_cast<const ColumnStruct&>(physical); + size_t index = 0; + const auto* child = find_child(schema, name, &index); + if (child == nullptr || index >= structure.tuple_size()) { + throw Exception(ErrorCode::CORRUPTION, "Parquet Variant {} has no physical child {}", + schema.name, name); + } + if (child_schema != nullptr) { + *child_schema = child; + } + return cell_at(structure.get_column(index), row); +} + +uint8_t decimal_width(int precision) { + if (precision <= 0 || precision > 38) { + throw Exception(ErrorCode::CORRUPTION, + "Parquet Variant decimal precision {} is outside [1, 38]", precision); + } + return precision <= 9 ? 4 : (precision <= 18 ? 8 : 16); +} + +uint8_t integer_width(const ParquetColumnSchema& schema, PrimitiveType type) { + if (schema.type_descriptor.is_unsigned_integer) { + throw Exception(ErrorCode::NOT_IMPLEMENTED_ERROR, + "Unsigned integers are not valid Parquet Variant typed values"); + } + if (schema.type_descriptor.integer_bit_width > 0) { + switch (schema.type_descriptor.integer_bit_width) { + case 8: + return 1; + case 16: + return 2; + case 32: + return 4; + case 64: + return 8; + default: + throw Exception(ErrorCode::CORRUPTION, "Invalid Parquet Variant integer width {}", + schema.type_descriptor.integer_bit_width); + } + } + switch (type) { + case TYPE_TINYINT: + return 1; + case TYPE_SMALLINT: + return 2; + case TYPE_INT: + return 4; + case TYPE_BIGINT: + return 8; + default: + throw Exception(ErrorCode::CORRUPTION, "Invalid Parquet Variant integer type {}", type); + } +} + +void append_typed_scalar(const ParquetColumnSchema& schema, const IColumn& column, size_t row, + VariantBatchBuilder::Row& builder) { + const PrimitiveType type = remove_nullable(schema.type)->get_primitive_type(); + switch (type) { + case TYPE_BOOLEAN: + builder.add_bool(assert_cast<const ColumnUInt8&>(column).get_data()[row] != 0); + return; + case TYPE_TINYINT: + builder.add_scalar( + VariantScalarRef::integer(assert_cast<const ColumnInt8&>(column).get_data()[row], + integer_width(schema, type))); + return; + case TYPE_SMALLINT: + builder.add_scalar( + VariantScalarRef::integer(assert_cast<const ColumnInt16&>(column).get_data()[row], + integer_width(schema, type))); + return; + case TYPE_INT: + builder.add_scalar( + VariantScalarRef::integer(assert_cast<const ColumnInt32&>(column).get_data()[row], + integer_width(schema, type))); + return; + case TYPE_BIGINT: + builder.add_scalar( + VariantScalarRef::integer(assert_cast<const ColumnInt64&>(column).get_data()[row], + integer_width(schema, type))); + return; + case TYPE_FLOAT: + builder.add_float(assert_cast<const ColumnFloat32&>(column).get_data()[row]); + return; + case TYPE_DOUBLE: + builder.add_double(assert_cast<const ColumnFloat64&>(column).get_data()[row]); + return; + case TYPE_DECIMAL128I: { + const auto value = assert_cast<const ColumnDecimal128V3&>(column).get_data()[row].value; + builder.add_decimal(value, static_cast<uint8_t>(schema.type_descriptor.decimal_scale), + decimal_width(schema.type_descriptor.decimal_precision)); + return; + } + case TYPE_TIMEV2: { + const double seconds = assert_cast<const ColumnTimeV2&>(column).get_data()[row]; + if (!std::isfinite(seconds) || + std::abs(seconds) > static_cast<double>(std::numeric_limits<int64_t>::max()) / 1e6) { + throw Exception(ErrorCode::CORRUPTION, "Invalid Parquet Variant TIME value"); + } + builder.add_time_ntz_micros(static_cast<int64_t>(std::llround(seconds * 1e6))); + return; + } + case TYPE_DATETIMEV2: { + if (schema.type_descriptor.time_unit == ParquetTimeUnit::NANOS) { + // Native DATETIMEV2 is microsecond based. Reject before returning a silently truncated + // value; a raw INT64 nanos decoder can be added independently. + throw Exception(ErrorCode::NOT_IMPLEMENTED_ERROR, + "Parquet Variant TIMESTAMP(NANOS) is not supported"); + } + const auto& value = assert_cast<const ColumnDateTimeV2&>(column).get_data()[row]; + builder.add_timestamp_micros( + variant_timestamp_micros(value, row, "Parquet Variant TIMESTAMP"), + schema.type_descriptor.timestamp_is_adjusted_to_utc); + return; + } + case TYPE_TIMESTAMPTZ: { + if (schema.type_descriptor.time_unit == ParquetTimeUnit::NANOS) { + throw Exception(ErrorCode::NOT_IMPLEMENTED_ERROR, + "Parquet Variant TIMESTAMP(NANOS) is not supported"); + } + const auto& value = assert_cast<const ColumnTimeStampTz&>(column).get_data()[row]; + builder.add_timestamp_micros( + variant_timestamp_micros(value, row, "Parquet Variant TIMESTAMP"), true); + return; + } + case TYPE_VARBINARY: { + const StringRef value = column.get_data_at(row); + if (!schema.type_descriptor.is_uuid) { + builder.add_binary(value); + return; + } + if (value.size != 16) { + throw Exception(ErrorCode::CORRUPTION, + "Parquet Variant UUID has {} bytes instead of 16", value.size); + } + std::array<uint8_t, 16> uuid {}; + std::memcpy(uuid.data(), value.data, uuid.size()); + builder.add_uuid(uuid); + return; + } + case TYPE_STRING: { + const StringRef value = column.get_data_at(row); + if (schema.type_descriptor.is_uuid) { + if (value.size != 16) { + throw Exception(ErrorCode::CORRUPTION, + "Parquet Variant UUID has {} bytes instead of 16", value.size); + } + std::array<uint8_t, 16> uuid {}; + std::memcpy(uuid.data(), value.data, uuid.size()); + builder.add_uuid(uuid); + } else if (schema.type_descriptor.is_string_annotation) { + builder.add_string(value); + } else { + builder.add_binary(value); + } + return; + } + default: + if (!is_supported_variant_typed_identity(type)) { + throw Exception(ErrorCode::NOT_IMPLEMENTED_ERROR, + "Parquet Variant typed value {} is not supported", + remove_nullable(schema.type)->get_name()); + } + dispatch_variant_typed_column( + column, type, [&]<PrimitiveType Type>(const auto& typed_column) { + with_variant_typed_scalar<Type>( + typed_column, row, + static_cast<uint8_t>(remove_nullable(schema.type)->get_scale()), + [&](const VariantScalarRef& scalar) { builder.add_scalar(scalar); }); + }); + } +} + +enum class WrapperContext { ROOT, ARRAY_ELEMENT, OBJECT_FIELD }; + +bool append_wrapper(const ParquetColumnSchema& schema, const IColumn& wrapper, size_t row, + VariantMetadataRef metadata, VariantBatchBuilder::Row& builder, + WrapperContext context); + +void append_typed_value(const ParquetColumnSchema& schema, const IColumn& column, size_t row, + VariantMetadataRef metadata, const VariantRef* residual, + VariantBatchBuilder::Row& builder) { + switch (schema.kind) { + case ParquetColumnSchemaKind::PRIMITIVE: + if (static_cast<bool>(residual)) { + throw Exception(ErrorCode::CORRUPTION, + "Parquet Variant scalar typed_value cannot have residual value bytes"); + } + append_typed_scalar(schema, column, row, builder); + return; + case ParquetColumnSchemaKind::STRUCT: { + if (static_cast<bool>(residual) && residual->basic_type() != VariantBasicType::OBJECT) { + throw Exception(ErrorCode::CORRUPTION, + "Parquet Variant object typed_value has non-object residual value"); + } + const auto& structure = assert_cast<const ColumnStruct&>(column); + if (structure.tuple_size() != schema.children.size()) { + throw Exception(ErrorCode::CORRUPTION, + "Parquet Variant object {} physical field count mismatch", schema.name); + } + auto object = builder.start_object(); + if (static_cast<bool>(residual)) { + for (uint32_t i = 0; i < residual->num_elements(); ++i) { + uint32_t field_id = 0; + const VariantRef child = residual->object_value_at(i, &field_id); + object.add_key(residual->metadata.key_at(field_id)); + builder.add_value(child); + } + } + for (size_t i = 0; i < schema.children.size(); ++i) { + const auto& child_schema = *schema.children[i]; + const Cell child = cell_at(structure.get_column(i), row); + if (child.is_null) { + // Shredded object fields are optional wrapper groups. A missing group means the + // key is absent, which differs from a present wrapper encoding a Variant null. + continue; + } + // A null/null wrapper means this object field is absent. Delay add_key until its + // presence is known so absent shredded fields do not turn into Variant nulls. + size_t value_index = 0; + const auto* value_schema = find_child(child_schema, "value", &value_index); + const auto& child_struct = assert_cast<const ColumnStruct&>(*child.column); + const bool value_present = value_schema != nullptr && + !cell_at(child_struct.get_column(value_index), row).is_null; + size_t typed_index = 0; + const auto* typed_schema = find_child(child_schema, "typed_value", &typed_index); + const bool typed_present = typed_schema != nullptr && + !cell_at(child_struct.get_column(typed_index), row).is_null; + if (!value_present && !typed_present) { + continue; + } + object.add_key(StringRef(child_schema.name)); + (void)append_wrapper(child_schema, *child.column, row, metadata, builder, + WrapperContext::OBJECT_FIELD); + } + object.finish(); + return; + } + case ParquetColumnSchemaKind::LIST: { + if (static_cast<bool>(residual)) { + throw Exception(ErrorCode::CORRUPTION, + "Parquet Variant array typed_value cannot have residual value bytes"); + } + if (schema.children.size() != 1) { + throw Exception(ErrorCode::CORRUPTION, + "Parquet Variant array {} has invalid element schema", schema.name); + } + const auto& array = assert_cast<const ColumnArray&>(column); + const size_t begin = array.offset_at(static_cast<ssize_t>(row)); + const size_t end = array.get_offsets()[row]; + auto scope = builder.start_array(); + for (size_t element = begin; element < end; ++element) { + const Cell cell = cell_at(array.get_data(), element); + if (cell.is_null) { + throw Exception(ErrorCode::CORRUPTION, + "Parquet Variant shredded array element wrapper is null"); + } + (void)append_wrapper(*schema.children[0], *cell.column, element, metadata, builder, + WrapperContext::ARRAY_ELEMENT); + } + scope.finish(); + return; + } + case ParquetColumnSchemaKind::MAP: + case ParquetColumnSchemaKind::VARIANT: + throw Exception(ErrorCode::CORRUPTION, "Invalid Parquet Variant typed_value schema {}", + schema.name); + } +} + +bool append_wrapper(const ParquetColumnSchema& schema, const IColumn& wrapper, size_t row, + VariantMetadataRef metadata, VariantBatchBuilder::Row& builder, + WrapperContext context) { + Cell value; + if (find_child(schema, "value", nullptr) != nullptr) { + value = struct_child_at(schema, wrapper, row, "value", nullptr); + } else { + value.is_null = true; + } + const ParquetColumnSchema* typed_schema = nullptr; + Cell typed; + if (find_child(schema, "typed_value", nullptr) != nullptr) { + typed = struct_child_at(schema, wrapper, row, "typed_value", &typed_schema); + } else { + typed.is_null = true; + } + + if (find_child(schema, "value", nullptr) == nullptr && typed_schema == nullptr) { + throw Exception(ErrorCode::CORRUPTION, + "Parquet Variant wrapper {} has neither value nor typed_value", + schema.name); + } + if (value.is_null && typed.is_null) { + if (context == WrapperContext::OBJECT_FIELD) { + return false; + } + if (context == WrapperContext::ARRAY_ELEMENT) { + throw Exception(ErrorCode::CORRUPTION, "Parquet Variant array element is missing"); + } + builder.add_null(); + return true; + } + + VariantRef residual {.metadata = metadata, .value = {}}; + if (!value.is_null) { + residual.value = value.column->get_data_at(row); + } + if (typed.is_null) { + builder.add_value(residual); + return true; + } + append_typed_value(*typed_schema, *typed.column, row, metadata, + value.is_null ? nullptr : &residual, builder); + return true; +} + +void encode_variant_range(const ParquetColumnSchema& schema, const IColumn& wrapper, + const ColumnNullable* outer_nullable, size_t begin, size_t end, + ColumnVariantV2& variants) { + try { + VariantBatchBuilder builder(VariantBatchBuilder::ReserveHint {.rows = end - begin}); + for (size_t row = begin; row < end; ++row) { + auto output_row = builder.begin_row(); + if (outer_nullable != nullptr && outer_nullable->get_null_map_data()[row] != 0) { + output_row.add_null(); + output_row.finish(); + continue; + } + const Cell metadata_cell = struct_child_at(schema, wrapper, row, "metadata", nullptr); + if (metadata_cell.is_null) { + throw Exception(ErrorCode::CORRUPTION, + "Parquet Variant {} has null metadata at row {}", schema.name, row); + } + const StringRef metadata_bytes = metadata_cell.column->get_data_at(row); + const VariantMetadataRef metadata {metadata_bytes.data, metadata_bytes.size}; + metadata.validate(); + (void)append_wrapper(schema, wrapper, row, metadata, output_row, WrapperContext::ROOT); + output_row.finish(); + } + VariantBatchBuilder batch = builder.finish_batch(); + variants.insert_encoded_batch(batch); + } catch (...) { + if (end - begin <= 1) { + throw; + } + // A single builder has one metadata dictionary. If heterogeneous file rows cannot fit in + // that dictionary, split without changing the destination column's already-valid batches. + // Corrupt input still reaches a one-row range and propagates its original exception. + const size_t middle = begin + (end - begin) / 2; + encode_variant_range(schema, wrapper, outer_nullable, begin, middle, variants); + encode_variant_range(schema, wrapper, outer_nullable, middle, end, variants); + } +} + +ColumnVariantV2::MutablePtr encode_variant_column(const ParquetColumnSchema& schema, + const IColumn& physical) { + if (schema.kind != ParquetColumnSchemaKind::VARIANT) { + throw Exception(ErrorCode::INVALID_ARGUMENT, "Parquet column {} is not Variant", + schema.name); + } + const auto* outer_nullable = check_and_get_column<ColumnNullable>(physical); + const IColumn& wrapper = + outer_nullable == nullptr ? physical : outer_nullable->get_nested_column(); + const auto& structure = assert_cast<const ColumnStruct&>(wrapper); + if (structure.tuple_size() != schema.children.size()) { + throw Exception(ErrorCode::CORRUPTION, "Parquet Variant {} physical field count mismatch", + schema.name); + } + + auto variants = ColumnVariantV2::create(); + constexpr size_t MAX_RECONSTRUCTION_BATCH_ROWS = 4096; + for (size_t begin = 0; begin < physical.size(); begin += MAX_RECONSTRUCTION_BATCH_ROWS) { + encode_variant_range(schema, wrapper, outer_nullable, begin, + std::min(physical.size(), begin + MAX_RECONSTRUCTION_BATCH_ROWS), + *variants); + } + return variants; +} + +std::unique_ptr<ParquetColumnSchema> clone_schema( + const ParquetColumnSchema& source, const format::LocalColumnIndex* projection = nullptr) { + auto result = std::make_unique<ParquetColumnSchema>(); + result->local_id = source.local_id; + result->parquet_field_id = source.parquet_field_id; + result->name = source.name; + result->type = source.type; + result->variant_physical_type = source.variant_physical_type; + result->leaf_column_id = source.leaf_column_id; + result->type_descriptor = source.type_descriptor; + result->kind = source.kind; + result->max_definition_level = source.max_definition_level; + result->max_repetition_level = source.max_repetition_level; + result->nullable_definition_level = source.nullable_definition_level; + result->definition_level = source.definition_level; + result->repetition_level = source.repetition_level; + result->repeated_ancestor_definition_level = source.repeated_ancestor_definition_level; + result->repeated_repetition_level = source.repeated_repetition_level; + const bool partial = format::is_partial_projection(projection); + result->children.reserve(partial ? projection->children.size() : source.children.size()); + if (partial) { + // NativeColumnReader emits a partial STRUCT in projection order, so the retained schema + // must use that same order or field names will address the wrong physical tuple element. + for (const auto& child_projection : projection->children) { + const auto child = std::ranges::find_if(source.children, [&](const auto& candidate) { + return candidate->local_id == child_projection.local_id(); + }); + DORIS_CHECK(child != source.children.end()); + result->children.push_back(clone_schema(**child, &child_projection)); + } + } else { + for (const auto& child : source.children) { + result->children.push_back(clone_schema(*child)); + } + } + return result; +} + +ColumnPtr unwrap_nullable(ColumnPtr column) { + if (const auto* nullable = check_and_get_column<ColumnNullable>(*column)) { + return nullable->get_nested_column_ptr(); + } + return column; +} + +ColumnPtr struct_child(const ParquetColumnSchema& schema, ColumnPtr column, std::string_view name, + const ParquetColumnSchema** child_schema) { + column = unwrap_nullable(std::move(column)); + const auto* structure = check_and_get_column<ColumnStruct>(*column); + if (structure == nullptr) { + return nullptr; + } + size_t index = 0; + const auto* child = find_child(schema, name, &index); + if (child == nullptr || index >= structure->tuple_size()) { + return nullptr; + } + if (child_schema != nullptr) { + *child_schema = child; + } + return structure->get_column_ptr(index); +} + +bool has_present_value(const ColumnPtr& column) { + if (const auto* nullable = check_and_get_column<ColumnNullable>(*column)) { + return std::ranges::any_of(nullable->get_null_map_data(), + [](uint8_t is_null) { return is_null == 0; }); + } + return !column->empty(); +} + +bool supports_direct_typed_variant_state(const ParquetColumnSchema& schema) { + if (schema.type == nullptr || schema.kind != ParquetColumnSchemaKind::PRIMITIVE) { + return false; + } + // ColumnVariantV2 typed state carries only a Doris type. Binary/UUID annotations, temporal + // units, and other Parquet-only identity must therefore reconstruct canonical Variant bytes. + switch (remove_nullable(schema.type)->get_primitive_type()) { + case TYPE_BOOLEAN: + case TYPE_TINYINT: + case TYPE_SMALLINT: + case TYPE_INT: + case TYPE_BIGINT: + case TYPE_FLOAT: + case TYPE_DOUBLE: + case TYPE_DECIMAL128I: + case TYPE_DATEV2: + return true; + default: + return false; + } +} + +bool same_data_type(const DataTypePtr& left, const DataTypePtr& right) { + return (!left && !right) || (left && right && left->equals(*right)); +} + +bool same_type_descriptor(const ParquetTypeDescriptor& left, const ParquetTypeDescriptor& right) { + return same_data_type(left.doris_type, right.doris_type) && + same_data_type(left.physical_doris_type, right.physical_doris_type) && + left.extra_type_info == right.extra_type_info && left.time_unit == right.time_unit && + left.physical_type == right.physical_type && + left.integer_bit_width == right.integer_bit_width && + left.decimal_precision == right.decimal_precision && + left.decimal_scale == right.decimal_scale && left.fixed_length == right.fixed_length && + left.is_unsigned_integer == right.is_unsigned_integer && + left.is_decimal == right.is_decimal && left.is_timestamp == right.is_timestamp && + left.timestamp_is_adjusted_to_utc == right.timestamp_is_adjusted_to_utc && + left.is_string_like == right.is_string_like && + left.is_string_annotation == right.is_string_annotation && + left.is_uuid == right.is_uuid && left.unsupported_reason == right.unsupported_reason; +} + +bool same_shredded_schema(const ParquetColumnSchema& left, const ParquetColumnSchema& right) { + if (left.name != right.name || left.kind != right.kind || + !same_data_type(left.type, right.type) || + !same_type_descriptor(left.type_descriptor, right.type_descriptor) || + left.children.size() != right.children.size()) { + return false; + } + for (size_t i = 0; i < left.children.size(); ++i) { + if (!same_shredded_schema(*left.children[i], *right.children[i])) { + return false; + } + } + return true; +} + +void append_compatible_column(IColumn& output, const IColumn& converted); +void validate_compatible_column(const IColumn& output, const IColumn& converted); + +class ParquetVariantShreddedState final : public VariantShreddedState { +public: + ParquetVariantShreddedState(std::shared_ptr<const ParquetColumnSchema> schema, + ColumnPtr physical, bool complete) + : _schema(std::move(schema)), _physical(std::move(physical)), _complete(complete) { + DORIS_CHECK(_schema != nullptr && static_cast<bool>(_physical)); + const ColumnPtr wrapper = unwrap_nullable(_physical); + const auto* structure = check_and_get_column<ColumnStruct>(*wrapper); + if (structure == nullptr || structure->tuple_size() != _schema->children.size()) { + throw Exception(ErrorCode::CORRUPTION, + "Parquet Variant {} physical field count mismatch", _schema->name); + } + } + + size_t size() const override { return _physical->size(); } + size_t byte_size() const override { + std::lock_guard lock(_materialization_lock); + return _physical->byte_size() + (_materialized ? _materialized->byte_size() : 0); + } + size_t allocated_bytes() const override { + std::lock_guard lock(_materialization_lock); + return _physical->allocated_bytes() + + (_materialized ? _materialized->allocated_bytes() : 0); + } + void sanity_check() const override { _physical->sanity_check(); } + + void for_each_subcolumn(const IColumn::ImutableColumnCallback& callback) const override { + callback(*_physical); + } + + std::shared_ptr<VariantShreddedState> filter(const IColumn::Filter& filter, + ssize_t result_size_hint) const override { + // Compact the decoded physical tree directly. In particular, a leaf-only projection has + // no metadata/value columns from which a canonical Variant could be reconstructed. + // The projection schema is immutable and reader-scoped, so derived selections share it + // instead of cloning the whole shredded tree for every filter operation. + return std::make_shared<ParquetVariantShreddedState>( + _schema, _physical->filter(filter, result_size_hint), _complete); + } + + std::shared_ptr<VariantShreddedState> select_range(size_t start, size_t length) const override { + return std::make_shared<ParquetVariantShreddedState>(_schema, _physical->cut(start, length), + _complete); + } + + std::shared_ptr<VariantShreddedState> select_indices( + const uint32_t* indices_begin, const uint32_t* indices_end) const override { + MutableColumnPtr selected = _physical->clone_empty(); + selected->insert_indices_from(*_physical, indices_begin, indices_end); + return std::make_shared<ParquetVariantShreddedState>(_schema, std::move(selected), + _complete); + } + + bool try_append(const VariantShreddedState& source) override { + const auto* parquet_source = dynamic_cast<const ParquetVariantShreddedState*>(&source); + if (parquet_source == nullptr || _complete != parquet_source->_complete || + !same_shredded_schema(*_schema, *parquet_source->_schema)) { + return false; + } + validate_compatible_column(*_physical, *parquet_source->_physical); + auto mutable_physical = IColumn::mutate(std::move(_physical)); + append_compatible_column(*mutable_physical, *parquet_source->_physical); + _physical = std::move(mutable_physical); + std::lock_guard lock(_materialization_lock); + _materialized.reset(); + return true; + } + + std::optional<VariantShreddedTypedValue> find_typed_value( + std::span<const VariantShreddedPathSegment> path) const override { + if (path.empty()) { + return std::nullopt; + } + + const ParquetColumnSchema* typed_schema = nullptr; + ColumnPtr typed = struct_child(*_schema, _physical, "typed_value", &typed_schema); + if (!typed || typed_schema->kind != ParquetColumnSchemaKind::STRUCT) { + return std::nullopt; + } + + for (size_t position = 0; position < path.size(); ++position) { + if (path[position].kind != VariantShreddedPathSegment::Kind::OBJECT_KEY) { + return std::nullopt; + } + + const std::string_view key(path[position].key.data, path[position].key.size); + const ParquetColumnSchema* wrapper_schema = nullptr; + ColumnPtr wrapper = struct_child(*typed_schema, typed, key, &wrapper_schema); + if (!wrapper) { + return std::nullopt; + } + + if (ColumnPtr residual = struct_child(*wrapper_schema, wrapper, "value", nullptr); + static_cast<bool>(residual) && has_present_value(residual)) { + // A residual value can contribute data to the same logical object. Reconstructing + // is required in that case; returning only the typed leaf would drop information. + return std::nullopt; + } + + typed = struct_child(*wrapper_schema, wrapper, "typed_value", &typed_schema); + if (!typed) { + return std::nullopt; + } + if (position + 1 == path.size()) { + if (typed_schema->kind != ParquetColumnSchemaKind::PRIMITIVE || + check_and_get_column<ColumnNullable>(*typed) == nullptr || + !supports_direct_typed_variant_state(*typed_schema)) { + return std::nullopt; + } + return VariantShreddedTypedValue {.column = std::move(typed), + .type = remove_nullable(typed_schema->type)}; + } + if (typed_schema->kind != ParquetColumnSchemaKind::STRUCT) { + return std::nullopt; + } + } + return std::nullopt; + } + + const ColumnVariantV2& materialized_column() const override { + std::lock_guard lock(_materialization_lock); + if (!_complete) { + throw Exception( + ErrorCode::INTERNAL_ERROR, + "A projected Parquet Variant can only serve its validated shredded leaves"); + } + if (!_materialized) { + _materialized = encode_variant_column(*_schema, *_physical); Review Comment: Fixed. Lazy reconstruction is charged to VariantReconstructionTime/VariantReconstructedRows, and direct-leaf plus categorized fallback counters were added. Reader tests assert direct-leaf and reconstruction profiles. ########## be/src/format_v2/parquet/parquet_statistics.cpp: ########## @@ -452,6 +455,322 @@ std::optional<format::LocalColumnId> file_column_id_by_block_position( return std::nullopt; } +enum class VariantComparisonOp { EQ, NE, LT, LE, GT, GE }; + +struct VariantShreddedPredicate { + int slot_index = -1; + std::vector<std::string> path; + DataTypePtr comparison_type; + DataTypePtr literal_type; + Field literal; + VariantComparisonOp op = VariantComparisonOp::EQ; +}; + +std::string callable_name(const VExprSPtr& expr) { + if (const auto function = std::dynamic_pointer_cast<VectorizedFnCall>(expr); + function != nullptr) { + return function->function_name(); + } + return expr == nullptr ? std::string {} : expr->expr_name(); +} + +std::optional<VariantComparisonOp> variant_comparison_op(std::string_view name) { + if (name == "eq") { + return VariantComparisonOp::EQ; + } + if (name == "ne") { + return VariantComparisonOp::NE; + } + if (name == "lt") { + return VariantComparisonOp::LT; + } + if (name == "le") { + return VariantComparisonOp::LE; + } + if (name == "gt") { + return VariantComparisonOp::GT; + } + if (name == "ge") { + return VariantComparisonOp::GE; + } + return std::nullopt; +} + +VariantComparisonOp reverse_variant_comparison(VariantComparisonOp op) { + switch (op) { + case VariantComparisonOp::EQ: + case VariantComparisonOp::NE: + return op; + case VariantComparisonOp::LT: + return VariantComparisonOp::GT; + case VariantComparisonOp::LE: + return VariantComparisonOp::GE; + case VariantComparisonOp::GT: + return VariantComparisonOp::LT; + case VariantComparisonOp::GE: + return VariantComparisonOp::LE; + } + __builtin_unreachable(); +} + +std::optional<std::pair<Field, DataTypePtr>> variant_literal(const VExprSPtr& expr) { + const auto literal = std::dynamic_pointer_cast<VLiteral>(expr); + if (literal == nullptr || !literal->get_column_ptr() || literal->get_column_ptr()->empty()) { + return std::nullopt; + } + Field value; + literal->get_column_ptr()->get(0, value); + if (value.is_null()) { + return std::nullopt; + } + return std::make_pair(std::move(value), literal->get_data_type()); +} + +std::optional<VariantShreddedPredicate> extract_variant_shredded_predicate( + const VExprContextSPtr& conjunct) { + if (conjunct == nullptr || conjunct->root() == nullptr || + conjunct->root()->get_num_children() != 2) { + return std::nullopt; + } + auto op = variant_comparison_op(callable_name(conjunct->root())); + if (!op.has_value()) { + return std::nullopt; + } + + VExprSPtr value_expr; + std::optional<std::pair<Field, DataTypePtr>> literal; + if ((literal = variant_literal(conjunct->root()->get_child(1))).has_value()) { + value_expr = conjunct->root()->get_child(0); + } else if ((literal = variant_literal(conjunct->root()->get_child(0))).has_value()) { + value_expr = conjunct->root()->get_child(1); + op = reverse_variant_comparison(*op); + } else { + return std::nullopt; + } + + const auto comparison_type = value_expr->data_type(); + while (value_expr->node_type() == TExprNodeType::CAST_EXPR && + value_expr->get_num_children() == 1) { + if (!expr_zonemap::data_types_compatible(value_expr->data_type(), comparison_type)) { + // Every removed cast must preserve the comparison domain. Otherwise bounds for the + // raw typed leaf could skip rows whose value changes in an intermediate narrowing cast. + return std::nullopt; + } + value_expr = value_expr->get_child(0); + } + + std::vector<std::string> reverse_path; + while (callable_name(value_expr) == "element_at" && value_expr->get_num_children() == 2) { + const auto key = variant_literal(value_expr->get_child(1)); + if (!key.has_value() || key->first.get_type() != TYPE_STRING) { + // Repeated array shredding has no single scalar page range, so only object keys are + // eligible for this file-level optimization. + return std::nullopt; + } + reverse_path.push_back(key->first.get<TYPE_STRING>()); + value_expr = value_expr->get_child(0); + } + const auto slot = std::dynamic_pointer_cast<VSlotRef>(value_expr); + if (slot == nullptr || reverse_path.empty() || comparison_type == nullptr || + remove_nullable(slot->data_type())->get_primitive_type() != TYPE_VARIANT || + !expr_zonemap::data_types_compatible(comparison_type, literal->second)) { + return std::nullopt; + } + std::ranges::reverse(reverse_path); + return VariantShreddedPredicate {.slot_index = slot->column_id(), + .path = std::move(reverse_path), + .comparison_type = comparison_type, + .literal_type = literal->second, + .literal = std::move(literal->first), + .op = *op}; +} + +bool has_variant_shredded_filter(const format::FileScanRequest& request) { + return std::ranges::any_of(request.conjuncts, [](const auto& conjunct) { + return extract_variant_shredded_predicate(conjunct).has_value(); + }); +} + +const ParquetColumnSchema* child_named(const ParquetColumnSchema& parent, std::string_view name) { + const auto it = std::ranges::find_if(parent.children, [&](const auto& child) { + return child != nullptr && child->name == name; + }); + return it == parent.children.end() ? nullptr : it->get(); +} + +struct ResolvedVariantShredding { + const ParquetColumnSchema* fallback_value = nullptr; + const ParquetColumnSchema* typed_value = nullptr; +}; + +bool metadata_cast_is_order_preserving(const DataTypePtr& source, const DataTypePtr& target) { + if (expr_zonemap::data_types_compatible(source, target)) { + return true; + } + const auto source_type = remove_nullable(source); + const auto target_type = remove_nullable(target); + const auto source_primitive = source_type->get_primitive_type(); + const auto target_primitive = target_type->get_primitive_type(); + // Metadata bounds may cross only exact widening domains. This mirrors the residual CAST while + // excluding rounding, overflow, and narrowing cases that could reverse a pruning decision. + if (source_primitive == TYPE_FLOAT && target_primitive == TYPE_DOUBLE) { + return true; + } + if (is_int(source_primitive) && source_primitive != TYPE_LARGEINT && + is_decimalv3(target_primitive)) { + const uint32_t required_integer_digits = source_primitive == TYPE_TINYINT ? 3 + : source_primitive == TYPE_SMALLINT ? 5 + : source_primitive == TYPE_INT ? 10 + : 19; + return target_type->get_precision() >= target_type->get_scale() && + target_type->get_precision() - target_type->get_scale() >= required_integer_digits; + } + if (is_decimalv3(source_primitive) && is_decimalv3(target_primitive)) { + const uint32_t source_integer_digits = + source_type->get_precision() - source_type->get_scale(); + const uint32_t target_integer_digits = + target_type->get_precision() - target_type->get_scale(); + return target_integer_digits >= source_integer_digits && + target_type->get_scale() >= source_type->get_scale(); + } + return false; +} + +std::optional<Field> cast_metadata_field(const Field& value, const DataTypePtr& source, + const DataTypePtr& target) { + if (expr_zonemap::data_types_compatible(source, target)) { + return value; + } + const auto source_type = remove_nullable(source); + const auto target_type = remove_nullable(target); + if (source_type->get_primitive_type() == TYPE_FLOAT && + target_type->get_primitive_type() == TYPE_DOUBLE) { + return Field::create_field<TYPE_DOUBLE>(static_cast<double>(value.get<TYPE_FLOAT>())); + } + try { + auto source_column = source_type->create_column(); + source_column->insert(value); + DataTypeSerDe::FormatOptions options = DataTypeSerDe::get_default_format_options(); + options.converted_from_string = true; + std::string text = source_type->to_string(*source_column, 0, options); + StringRef input(text.data(), text.size()); + auto target_column = target_type->create_column(); + if (!target_type->get_serde() + ->from_string_strict_mode(input, *target_column, options) + .ok() || + target_column->size() != 1) { + return std::nullopt; + } + Field result; + target_column->get(0, result); + return result; + } catch (...) { + return std::nullopt; + } +} + +std::optional<ParquetColumnStatistics> normalize_variant_statistics( + const VariantShreddedPredicate& predicate, const ParquetColumnSchema& typed_value, + const ParquetColumnStatistics& statistics) { + if (!statistics.has_min_max || + expr_zonemap::data_types_compatible(typed_value.type, predicate.comparison_type)) { + return statistics; + } + auto min_value = + cast_metadata_field(statistics.min_value, typed_value.type, predicate.comparison_type); + auto max_value = + cast_metadata_field(statistics.max_value, typed_value.type, predicate.comparison_type); + if (!min_value.has_value() || !max_value.has_value()) { + return std::nullopt; + } + auto normalized = statistics; + normalized.min_value = std::move(*min_value); + normalized.max_value = std::move(*max_value); + return normalized; +} + +std::optional<ResolvedVariantShredding> resolve_variant_shredding( + const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema, + const format::FileScanRequest& request, const VariantShreddedPredicate& predicate) { + const auto local_id = file_column_id_by_block_position(request, predicate.slot_index); + if (!local_id.has_value() || local_id->value() < 0 || + local_id->value() >= static_cast<int>(file_schema.size())) { + return std::nullopt; + } + const ParquetColumnSchema* wrapper = file_schema[local_id->value()].get(); + if (wrapper == nullptr || wrapper->kind != ParquetColumnSchemaKind::VARIANT) { + return std::nullopt; + } + for (const auto& component : predicate.path) { + const auto* typed_object = child_named(*wrapper, "typed_value"); + if (typed_object == nullptr || typed_object->kind != ParquetColumnSchemaKind::STRUCT) { + return std::nullopt; + } + wrapper = child_named(*typed_object, component); + if (wrapper == nullptr || wrapper->kind != ParquetColumnSchemaKind::STRUCT) { + return std::nullopt; + } + } + const auto* fallback = child_named(*wrapper, "value"); + const auto* typed = child_named(*wrapper, "typed_value"); + const auto typed_primitive = typed == nullptr || typed->type == nullptr + ? INVALID_TYPE + : remove_nullable(typed->type)->get_primitive_type(); + if (fallback == nullptr || typed == nullptr || + fallback->kind != ParquetColumnSchemaKind::PRIMITIVE || + typed->kind != ParquetColumnSchemaKind::PRIMITIVE || typed->max_repetition_level != 0 || + // Parquet float statistics do not prove that a page contains no NaN. Min/max pruning in + // the presence of NaN is not order preserving, so keep those pages until such proof exists. + typed_primitive == TYPE_FLOAT || typed_primitive == TYPE_DOUBLE || + !metadata_cast_is_order_preserving(typed->type, predicate.comparison_type) || Review Comment: Fixed. UUID and unannotated raw-binary typed leaves are excluded from Variant footer and page-index pruning because their physical bounds differ from residual STRING semantics. Differential tests cover both pruning levels. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
