csun5285 commented on code in PR #65561: URL: https://github.com/apache/doris/pull/65561#discussion_r3649664797
########## be/src/core/column/variant_v2/column_variant_v2.h: ########## @@ -0,0 +1,283 @@ +// 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. + +#pragma once + +#include <array> +#include <limits> +#include <span> +#include <string> + +#include "common/exception.h" +#include "core/assert_cast.h" +#include "core/column/column.h" +#include "core/column/column_const.h" +#include "core/column/variant_v2/column_variant_v2_typed_column.h" +#include "core/custom_allocator.h" +#include "core/data_type/data_type.h" +#include "core/value/variant/variant_parquet_encoding.h" +#include "core/value/variant/variant_value.h" + +namespace doris { + +class DataTypeVariantV2SerDe; +class VariantBatchBuilder; + +// ColumnVariantV2 stores a whole column in exactly one state: encoded Variant bytes or one nullable +// typed scalar column. Mixed operations materialize the typed state as encoded bytes on demand. +class ColumnVariantV2 final : public COWHelper<IColumn, ColumnVariantV2> { +public: + struct EncodedDataView { + StringRef metadata_bytes; + std::span<const uint32_t> metadata_offsets; + std::span<const uint32_t> meta_ids; + StringRef value_bytes; + std::span<const uint32_t> value_offsets; + }; + + // Borrowed immutable adapter for whole-column E/T readers. The source column owns every + // referenced column, type, and byte; any structural mutation invalidates this view. Encoded + // bytes have already been validated at their insertion or deserialization boundary. + class ReadView { + public: + bool is_typed() const noexcept { return _typed_state; } + size_t size() const noexcept; + size_t metadata_count() const noexcept; + uint32_t metadata_id_at(size_t row) const; + VariantMetadataRef metadata_at(uint32_t id) const; + VariantRef value_at(size_t row) const; + const IColumn& typed_column() const; + const DataTypePtr& typed_type() const; + + private: + friend class ColumnVariantV2; + ReadView(const IColumn* metadatas, const IColumn* metadata_ids, const IColumn* values); + ReadView(const IColumn* typed, const DataTypePtr* typed_type); + + bool _typed_state = false; + const IColumn* _metadatas = nullptr; + const IColumn* _metadata_ids = nullptr; + const IColumn* _values = nullptr; + const IColumn* _typed = nullptr; + const DataTypePtr* _typed_type = nullptr; + }; + +#ifdef BE_TEST + // Narrow unit-test seam for encoded-state invariant coverage. + struct TestAccess { + static void replace_encoded_subcolumn(ColumnVariantV2& column, size_t index, + ColumnPtr replacement); + }; +#endif + + // The input must be an exact, non-Const ColumnNullable whose nested column matches the + // non-nullable supported scalar type. + static MutablePtr create_typed(ColumnPtr column, DataTypePtr scalar_type); + + bool is_typed() const noexcept { return _typed != nullptr; } + const IColumn& typed_column() const; + const DataTypePtr& typed_type() const; + void ensure_encoded(); + ReadView read_view() const; + + std::string get_name() const override; + size_t size() const override; + size_t byte_size() const override; + size_t allocated_bytes() const override; + bool has_enough_capacity(const IColumn& src) const override; + bool is_variable_length() const override { return true; } + bool structure_equals(const IColumn& rhs) const override; + + void sanity_check() const override; + void for_each_subcolumn(ColumnCallback callback) const override; + void clear() override; + void finalize() override {} + + // Validates the borrowed buffer/offset/id structure, then appends codec-validated encoded rows + // without retaining any input pointer. Offsets use the ColumnString uint32 domain and start at + // zero. Empty meta_ids is the compact representation for a batch whose rows all use its single + // metadata blob. Input buffers must not alias this column; use insert_range_from for that case. + void insert_encoded_rows(const EncodedDataView& data); + + // Direct trusted codec adapter. VariantBatchBuilder already produces canonical metadata, + // validated values, and ColumnString-compatible uint32 offsets, so this path copies its buffers + // without validating the encoded tree a second time. + void insert_encoded_batch(const VariantBatchBuilder& block); + + // The returned view borrows this column's metadata and value buffers. Any structural mutation, + // including insert, clear, COW mutation, or future row transformations, may invalidate it. + VariantRef get_value_ref(size_t row) const; + + Field operator[](size_t row) const override; + void get(size_t row, Field& result) const override; + void insert(const Field& field) override; + void insert_default() override; + void insert_many_defaults(size_t length) override; + + void insert_from(const IColumn& src, size_t row) override; + void insert_range_from(const IColumn& src, size_t start, size_t length) override; + void insert_indices_from(const IColumn& src, const uint32_t* indices_begin, + const uint32_t* indices_end) override; + void pop_back(size_t length) override; + + StringRef get_data_at(size_t row) const override; + void insert_data(const char* pos, size_t length) override; + StringRef serialize_value_into_arena(size_t row, Arena& arena, + const char*& begin) const override; + const char* deserialize_and_insert_from_arena(const char* pos) override; + size_t serialize_size_at(size_t row) const override; + size_t serialize_impl(char* pos, size_t row) const override; + size_t deserialize_impl(const char* pos) override; + size_t get_max_row_byte_size() const override; + void serialize(StringRef* keys, size_t num_rows) const override; + void deserialize(StringRef* keys, size_t num_rows) override; + + void update_hash_with_value(size_t row, SipHash& hash) const override; + void update_hashes_with_value(uint64_t* __restrict hashes, + const uint8_t* __restrict null_data) const override; + void update_xxHash_with_value(size_t start, size_t end, uint64_t& hash, + const uint8_t* __restrict null_data) const override; + void update_crcs_with_value(uint32_t* __restrict hashes, PrimitiveType type, uint32_t rows, + uint32_t offset, + const uint8_t* __restrict null_data) const override; + void update_crc_with_value(size_t start, size_t end, uint32_t& hash, + const uint8_t* __restrict null_data) const override; + void update_crc32c_batch(uint32_t* __restrict hashes, + const uint8_t* __restrict null_map) const override; + void update_crc32c_single(size_t start, size_t end, uint32_t& hash, + const uint8_t* __restrict null_map) const override; + void replace_column_null_data(const uint8_t* __restrict null_map) override; + + ColumnPtr filter(const Filter& filter, ssize_t result_size_hint) const override; + size_t filter(const Filter& filter) override; + MutableColumnPtr permute(const Permutation& permutation, size_t limit) const override; + MutableColumnPtr clone_resized(size_t size) const override; + void resize(size_t size) override; + + void get_permutation(bool reverse, size_t limit, int nan_direction_hint, HybridSorter& sorter, + Permutation& result) const override; + void replace_column_data(const IColumn& rhs, size_t row, size_t self_row = 0) override; + +private: + friend class COWHelper<IColumn, ColumnVariantV2>; + friend class DataTypeVariantV2SerDe; + + ColumnVariantV2(); + ColumnVariantV2(const ColumnVariantV2& other); + + uint32_t _find_or_insert_metadata(StringRef metadata); + void _adopt_state_from(ColumnVariantV2& replacement); + void _detach_metadata_for_write(); + void _check_invariants() const; + void mutate_subcolumns() override; + + // Encoded state: each row owns a value and references one deduplicated metadata blob. The + // uint32 id costs four bytes per encoded row, but avoids repeating object-key metadata and + // gives canonical comparison, hashing, subpath lookup, and binary SerDe O(1) schema access. + // It is required because valid external Variant rows may use different metadata dictionaries. + IColumn::WrappedPtr _metadatas; Review Comment: 这个考虑用 hashmap 实现？否则获取这一行的 meta 都是O（N) ########## be/src/exprs/function/cast/variant_v2/cast_variant_to_string.cpp: ########## @@ -0,0 +1,255 @@ +// 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 <algorithm> +#include <utility> + +#include "core/assert_cast.h" +#include "core/column/column_string.h" +#include "core/column/variant_v2/column_variant_v2.h" +#include "core/custom_allocator.h" +#include "core/data_type/data_type_string.h" +#include "core/string_buffer.hpp" +#include "exprs/function/cast/variant_v2/cast_variant_v2_internal.h" +#include "exprs/function/parse/variant_string_parse.h" +#include "exprs/function_context.h" +#include "runtime/runtime_state.h" + +namespace doris::CastWrapper::variant_v2_internal { +namespace { + +VariantJsonFormatOptions json_options(FunctionContext* context) { Review Comment: 这个和 cast_variant_to_jsonb 里面的json_options 什么区别 ########## be/src/core/column/variant_v2/column_variant_v2_typed_column.cpp: ########## @@ -0,0 +1,109 @@ +// 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 "core/column/variant_v2/column_variant_v2_typed_column.h" + +#include "core/data_type/data_type_string.h" + +namespace doris::column_variant_v2_internal { +namespace detail { + +size_t format_int128(__int128 value, char* const output) noexcept { + std::array<char, 39> reversed {}; + size_t digits = 0; + unsigned __int128 remaining = unsigned_magnitude(value); + do { + reversed[digits++] = static_cast<char>('0' + remaining % 10); + remaining /= 10; + } while (remaining != 0); + + size_t position = 0; + if (value < 0) { + output[position++] = '-'; + } + while (digits != 0) { + output[position++] = reversed[--digits]; + } + return position; +} + +} // namespace detail + +bool is_supported_typed_identity(PrimitiveType type) { Review Comment: 这个和 is_supported_scalar_source 区别是啥，这两个还不一样，都是表示从doris 的scalar 类型转成 parquet varaint 的scalar 类型？ ########## be/src/core/value/variant/variant_batch_builder.cpp: ########## @@ -0,0 +1,2147 @@ +// 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 "core/value/variant/variant_batch_builder.h" + +#include <parallel_hashmap/phmap.h> + +#include <algorithm> +#include <bit> +#include <cstdint> +#include <cstring> +#include <deque> +#include <functional> +#include <limits> +#include <ranges> +#include <string> +#include <utility> + +#include "common/exception.h" +#include "core/custom_allocator.h" +#include "core/pod_array.h" +#include "core/value/variant/variant_parquet_encoding.h" +#include "util/utf8_check.h" + +namespace doris { +namespace { + +constexpr uint32_t INVALID_INDEX = std::numeric_limits<uint32_t>::max(); +constexpr uint64_t FINISHED_ROW_GENERATION = std::numeric_limits<uint64_t>::max(); +constexpr uint64_t ABORTED_ROW_GENERATION = FINISHED_ROW_GENERATION - 1; +constexpr size_t INLINE_SCALAR_CAPACITY = sizeof(uint32_t); +constexpr size_t SMALL_OBJECT_SORT_THRESHOLD = 16; + +constexpr unsigned __int128 max_decimal38() { + unsigned __int128 value = 1; + for (uint8_t digit = 0; digit < 38; ++digit) { + value *= 10; + } + return value - 1; +} + +constexpr unsigned __int128 MAX_DECIMAL38 = max_decimal38(); + +template <typename Container, typename Less> +void sort_object_entries(Container& entries, size_t begin, size_t count, Less less) { + if (count < 2) { + return; + } + if (count <= SMALL_OBJECT_SORT_THRESHOLD) { + for (size_t index = begin + 1; index < begin + count; ++index) { + typename Container::value_type current = entries[index]; + size_t insertion = index; + while (insertion > begin && less(current, entries[insertion - 1])) { + entries[insertion] = entries[insertion - 1]; + --insertion; + } + entries[insertion] = current; + } + return; + } + std::sort(entries.begin() + begin, entries.begin() + begin + count, less); +} + +unsigned __int128 magnitude(__int128 value) { + const auto unsigned_value = static_cast<unsigned __int128>(value); + return value < 0 ? ~unsigned_value + 1 : unsigned_value; +} + +uint8_t minimum_unsigned_width(uint64_t value) { + if (value <= std::numeric_limits<uint8_t>::max()) { + return 1; + } + if (value <= std::numeric_limits<uint16_t>::max()) { + return 2; + } + if (value <= 0xFFFFFFU) { + return 3; + } + return 4; +} + +void write_unsigned(char*& output, uint64_t value, uint8_t width) noexcept { Review Comment: write_scalar_unsigned 这俩一样？ ########## be/src/core/value/variant/variant_field.cpp: ########## @@ -0,0 +1,406 @@ +// 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 "core/value/variant/variant_field.h" + +#include <algorithm> +#include <cstring> +#include <limits> +#include <utility> +#include <vector> + +#include "common/exception.h" +#include "core/value/variant/variant_parquet_encoding.h" +#include "util/utf8_check.h" + +namespace doris { +namespace { + +constexpr size_t METADATA_SIZE_PREFIX = sizeof(uint32_t); +constexpr unsigned __int128 DECIMAL4_MAX = 999'999'999; +constexpr unsigned __int128 DECIMAL8_MAX = 999'999'999'999'999'999; + +constexpr unsigned __int128 max_decimal38() { Review Comment: 这个似乎多次出现了 ########## be/src/exprs/function/cast/variant_v2/cast_variant_to_scalar.cpp: ########## @@ -0,0 +1,556 @@ +// 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 <cctz/time_zone.h> + +#include <algorithm> +#include <array> +#include <cstdint> +#include <memory> +#include <utility> + +#include "common/exception.h" +#include "core/assert_cast.h" +#include "core/block/block.h" +#include "core/column/column_decimal.h" +#include "core/column/column_string.h" +#include "core/column/column_vector.h" +#include "core/column/variant_v2/column_variant_v2.h" +#include "core/custom_allocator.h" +#include "core/data_type/data_type_date_or_datetime_v2.h" +#include "core/data_type/data_type_decimal.h" +#include "core/data_type/data_type_nullable.h" +#include "core/data_type/data_type_number.h" +#include "core/data_type/data_type_string.h" +#include "core/data_type/data_type_timestamptz.h" +#include "core/value/timestamptz_value.h" +#include "core/value/vdatetime_value.h" +#include "exprs/function/cast/cast_base.h" +#include "exprs/function/cast/variant_v2/cast_variant_v2_internal.h" +#include "exprs/function_context.h" + +namespace doris::CastWrapper::variant_v2_internal { +namespace { + +constexpr size_t DECIMAL_SCALE_COUNT = 39; + +enum GroupIndex : size_t { + INVALID_GROUP = 0, + BOOL_GROUP = INVALID_GROUP + 1, + INT_GROUP = BOOL_GROUP + 1, + FLOAT_GROUP = INT_GROUP + 1, + DOUBLE_GROUP = FLOAT_GROUP + 1, + DECIMAL_GROUP_BEGIN = DOUBLE_GROUP + 1, + DATE_GROUP = DECIMAL_GROUP_BEGIN + DECIMAL_SCALE_COUNT, + TIMESTAMP_NTZ_GROUP = DATE_GROUP + 1, + TIMESTAMP_TZ_GROUP = TIMESTAMP_NTZ_GROUP + 1, + STRING_GROUP = TIMESTAMP_TZ_GROUP + 1, + GROUP_COUNT = STRING_GROUP + 1, +}; + +struct ScalarGroup { + DataTypePtr type; + MutableColumnPtr values; + DorisVector<size_t> source_rows; +}; + +using ScalarGroups = std::array<ScalarGroup, GROUP_COUNT>; + +template <typename TypeFactory> +ScalarGroup& initialize_group(ScalarGroups& groups, size_t index, TypeFactory&& make_type) { + ScalarGroup& group = groups[index]; + if (!group.values) { + group.type = std::forward<TypeFactory>(make_type)(); + group.values = group.type->create_column(); + } + return group; +} + +void append_invalid(ScalarGroups& groups, size_t row) { + groups[INVALID_GROUP].source_rows.push_back(row); +} + +std::pair<int64_t, uint32_t> split_epoch_micros(int64_t micros) { + constexpr int64_t MICROS_PER_SECOND = 1'000'000; + int64_t seconds = micros / MICROS_PER_SECOND; + int64_t fraction = micros % MICROS_PER_SECOND; + if (fraction < 0) { + --seconds; + fraction += MICROS_PER_SECOND; + } + return {seconds, static_cast<uint32_t>(fraction)}; +} + +int64_t nanos_to_micros(int64_t nanos) { + int64_t micros = nanos / 1000; + if (nanos % 1000 < 0) { + --micros; + } + return micros; +} + +bool epoch_micros_to_civil(int64_t micros, cctz::civil_second* civil, uint32_t* fraction) { + const auto [seconds, micros_fraction] = split_epoch_micros(micros); + const auto lookup = + cctz::utc_time_zone().lookup(cctz::time_point<cctz::seconds>(cctz::seconds(seconds))); + if (lookup.cs.year() < 1 || lookup.cs.year() > 9999) { + return false; + } + *civil = lookup.cs; + *fraction = micros_fraction; + return true; +} + +void append_bool(ScalarGroups& groups, size_t row, bool value) { + auto& group = + initialize_group(groups, BOOL_GROUP, [] { return std::make_shared<DataTypeBool>(); }); + assert_cast<ColumnUInt8&>(*group.values).insert_value(value); + group.source_rows.push_back(row); +} + +void append_int(ScalarGroups& groups, size_t row, int64_t value) { + auto& group = + initialize_group(groups, INT_GROUP, [] { return std::make_shared<DataTypeInt64>(); }); + assert_cast<ColumnInt64&>(*group.values).insert_value(value); + group.source_rows.push_back(row); +} + +void append_float(ScalarGroups& groups, size_t row, float value) { + auto& group = initialize_group(groups, FLOAT_GROUP, + [] { return std::make_shared<DataTypeFloat32>(); }); + assert_cast<ColumnFloat32&>(*group.values).insert_value(value); + group.source_rows.push_back(row); +} + +void append_double(ScalarGroups& groups, size_t row, double value) { + auto& group = initialize_group(groups, DOUBLE_GROUP, + [] { return std::make_shared<DataTypeFloat64>(); }); + assert_cast<ColumnFloat64&>(*group.values).insert_value(value); + group.source_rows.push_back(row); +} + +void append_decimal(ScalarGroups& groups, size_t row, VariantDecimal value) { + if (value.scale >= DECIMAL_SCALE_COUNT) { + append_invalid(groups, row); + return; + } + const size_t index = DECIMAL_GROUP_BEGIN + value.scale; + auto& group = initialize_group(groups, index, [scale = value.scale] { + return std::make_shared<DataTypeDecimal128>(38, scale); + }); + assert_cast<ColumnDecimal128V3&>(*group.values).insert_value(Decimal128V3 {value.unscaled}); + group.source_rows.push_back(row); +} + +void append_date(ScalarGroups& groups, size_t row, int32_t days_since_epoch) { + const cctz::civil_day civil = cctz::civil_day(1970, 1, 1) + days_since_epoch; + if (civil.year() < 1 || civil.year() > 9999) { + append_invalid(groups, row); + return; + } + DateV2Value<DateV2ValueType> value; + value.unchecked_set_time(static_cast<uint16_t>(civil.year()), + static_cast<uint8_t>(civil.month()), static_cast<uint8_t>(civil.day()), + 0, 0, 0); + auto& group = + initialize_group(groups, DATE_GROUP, [] { return std::make_shared<DataTypeDateV2>(); }); + assert_cast<ColumnDateV2&>(*group.values).insert_value(value); + group.source_rows.push_back(row); +} + +void append_timestamp(ScalarGroups& groups, size_t row, int64_t micros, bool utc_adjusted) { + cctz::civil_second civil; + uint32_t fraction = 0; + if (!epoch_micros_to_civil(micros, &civil, &fraction)) { + append_invalid(groups, row); + return; + } + if (utc_adjusted) { + TimestampTzValue value; + value.unchecked_set_time( + static_cast<uint16_t>(civil.year()), static_cast<uint8_t>(civil.month()), + static_cast<uint8_t>(civil.day()), static_cast<uint8_t>(civil.hour()), + static_cast<uint8_t>(civil.minute()), static_cast<uint8_t>(civil.second()), + fraction); + auto& group = initialize_group(groups, TIMESTAMP_TZ_GROUP, + [] { return std::make_shared<DataTypeTimeStampTz>(6); }); + assert_cast<ColumnTimeStampTz&>(*group.values).insert_value(value); + group.source_rows.push_back(row); + return; + } + DateV2Value<DateTimeV2ValueType> value; + value.unchecked_set_time( + static_cast<uint16_t>(civil.year()), static_cast<uint8_t>(civil.month()), + static_cast<uint8_t>(civil.day()), static_cast<uint8_t>(civil.hour()), + static_cast<uint8_t>(civil.minute()), static_cast<uint8_t>(civil.second()), fraction); + auto& group = initialize_group(groups, TIMESTAMP_NTZ_GROUP, + [] { return std::make_shared<DataTypeDateTimeV2>(6); }); + assert_cast<ColumnDateTimeV2&>(*group.values).insert_value(value); + group.source_rows.push_back(row); +} + +void append_string(ScalarGroups& groups, size_t row, StringRef value) { + auto& group = initialize_group(groups, STRING_GROUP, + [] { return std::make_shared<DataTypeString>(); }); + assert_cast<ColumnString&>(*group.values).insert_data(value.data, value.size); + group.source_rows.push_back(row); +} + +void classify_value(ScalarGroups& groups, size_t row, VariantRef value, bool forced_null) { + if (forced_null) { + append_invalid(groups, row); + return; + } + switch (value.basic_type()) { + case VariantBasicType::SHORT_STRING: + append_string(groups, row, value.get_string()); + return; + case VariantBasicType::OBJECT: + case VariantBasicType::ARRAY: + append_invalid(groups, row); + return; + case VariantBasicType::PRIMITIVE: + break; + } + + switch (value.primitive_id()) { + case VariantPrimitiveId::NULL_VALUE: + case VariantPrimitiveId::BINARY: + case VariantPrimitiveId::TIME_NTZ_MICROS: + case VariantPrimitiveId::UUID: + append_invalid(groups, row); + return; + case VariantPrimitiveId::TRUE_VALUE: + append_bool(groups, row, true); + return; + case VariantPrimitiveId::FALSE_VALUE: + append_bool(groups, row, false); + return; + case VariantPrimitiveId::INT8: + case VariantPrimitiveId::INT16: + case VariantPrimitiveId::INT32: + case VariantPrimitiveId::INT64: + append_int(groups, row, value.get_int()); + return; + case VariantPrimitiveId::FLOAT: + append_float(groups, row, value.get_float()); + return; + case VariantPrimitiveId::DOUBLE: + append_double(groups, row, value.get_double()); + return; + case VariantPrimitiveId::DECIMAL4: + case VariantPrimitiveId::DECIMAL8: + case VariantPrimitiveId::DECIMAL16: + append_decimal(groups, row, value.get_decimal()); + return; + case VariantPrimitiveId::DATE: + append_date(groups, row, value.get_date()); + return; + case VariantPrimitiveId::TIMESTAMP_MICROS: + append_timestamp(groups, row, value.get_timestamp_micros(), true); + return; + case VariantPrimitiveId::TIMESTAMP_NTZ_MICROS: + append_timestamp(groups, row, value.get_timestamp_ntz_micros(), false); + return; + case VariantPrimitiveId::TIMESTAMP_NANOS: + append_timestamp(groups, row, nanos_to_micros(value.get_timestamp_nanos()), true); + return; + case VariantPrimitiveId::TIMESTAMP_NTZ_NANOS: + append_timestamp(groups, row, nanos_to_micros(value.get_timestamp_ntz_nanos()), false); + return; + case VariantPrimitiveId::STRING: + append_string(groups, row, value.get_string()); + return; + } + throw Exception(ErrorCode::CORRUPTION, "Unknown Variant primitive id"); +} + +ColumnPtr as_nullable(ColumnPtr column, size_t rows) { Review Comment: 和make_nullable 功能重复了 ########## be/src/core/column/variant_v2/column_variant_v2_typed_column.h: ########## @@ -0,0 +1,300 @@ +// 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. + +#pragma once + +#include <algorithm> +#include <array> +#include <cstdint> +#include <limits> +#include <string_view> +#include <utility> + +#include "common/check.h" +#include "common/exception.h" +#include "core/assert_cast.h" +#include "core/column/column_decimal.h" +#include "core/column/column_nullable.h" +#include "core/column/column_string.h" +#include "core/column/column_vector.h" +#include "core/data_type/data_type.h" +#include "core/value/variant/variant_batch_builder.h" +#include "core/value/variant/variant_canonical.h" +#include "exec/common/format_ip.h" + +namespace doris::column_variant_v2_internal { + +bool is_supported_typed_identity(PrimitiveType type); +bool exact_typed_identity(const DataTypePtr& left, const DataTypePtr& right); +void validate_typed_decimal_scale(const IColumn& nested, PrimitiveType type, uint32_t scale); + +namespace detail { + +constexpr unsigned __int128 max_decimal38() { + unsigned __int128 value = 1; + for (uint8_t digit = 0; digit < 38; ++digit) { + value *= 10; + } + return value - 1; +} + +constexpr unsigned __int128 MAX_DECIMAL38 = max_decimal38(); Review Comment: 这个 variant_canonical.cpp 也有，variant_batch_builder.cpp 也有 ########## be/src/core/column/variant_v2/column_variant_v2_typed_column.h: ########## @@ -0,0 +1,300 @@ +// 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. + +#pragma once + +#include <algorithm> +#include <array> +#include <cstdint> +#include <limits> +#include <string_view> +#include <utility> + +#include "common/check.h" +#include "common/exception.h" +#include "core/assert_cast.h" +#include "core/column/column_decimal.h" +#include "core/column/column_nullable.h" +#include "core/column/column_string.h" +#include "core/column/column_vector.h" +#include "core/data_type/data_type.h" +#include "core/value/variant/variant_batch_builder.h" +#include "core/value/variant/variant_canonical.h" +#include "exec/common/format_ip.h" + +namespace doris::column_variant_v2_internal { + +bool is_supported_typed_identity(PrimitiveType type); +bool exact_typed_identity(const DataTypePtr& left, const DataTypePtr& right); +void validate_typed_decimal_scale(const IColumn& nested, PrimitiveType type, uint32_t scale); + +namespace detail { + +constexpr unsigned __int128 max_decimal38() { + unsigned __int128 value = 1; + for (uint8_t digit = 0; digit < 38; ++digit) { + value *= 10; + } + return value - 1; +} + +constexpr unsigned __int128 MAX_DECIMAL38 = max_decimal38(); +constexpr int64_t MICROS_PER_SECOND = 1'000'000; +constexpr int64_t SECONDS_PER_DAY = 86'400; + +inline unsigned __int128 unsigned_magnitude(__int128 value) noexcept { Review Comment: 这个也是在多个文件出现 ########## be/src/exprs/function/cast/variant_v2/cast_variant_to_scalar.cpp: ########## @@ -0,0 +1,556 @@ +// 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 <cctz/time_zone.h> + +#include <algorithm> +#include <array> +#include <cstdint> +#include <memory> +#include <utility> + +#include "common/exception.h" +#include "core/assert_cast.h" +#include "core/block/block.h" +#include "core/column/column_decimal.h" +#include "core/column/column_string.h" +#include "core/column/column_vector.h" +#include "core/column/variant_v2/column_variant_v2.h" +#include "core/custom_allocator.h" +#include "core/data_type/data_type_date_or_datetime_v2.h" +#include "core/data_type/data_type_decimal.h" +#include "core/data_type/data_type_nullable.h" +#include "core/data_type/data_type_number.h" +#include "core/data_type/data_type_string.h" +#include "core/data_type/data_type_timestamptz.h" +#include "core/value/timestamptz_value.h" +#include "core/value/vdatetime_value.h" +#include "exprs/function/cast/cast_base.h" +#include "exprs/function/cast/variant_v2/cast_variant_v2_internal.h" +#include "exprs/function_context.h" + +namespace doris::CastWrapper::variant_v2_internal { +namespace { + +constexpr size_t DECIMAL_SCALE_COUNT = 39; + +enum GroupIndex : size_t { + INVALID_GROUP = 0, + BOOL_GROUP = INVALID_GROUP + 1, + INT_GROUP = BOOL_GROUP + 1, + FLOAT_GROUP = INT_GROUP + 1, + DOUBLE_GROUP = FLOAT_GROUP + 1, + DECIMAL_GROUP_BEGIN = DOUBLE_GROUP + 1, + DATE_GROUP = DECIMAL_GROUP_BEGIN + DECIMAL_SCALE_COUNT, + TIMESTAMP_NTZ_GROUP = DATE_GROUP + 1, + TIMESTAMP_TZ_GROUP = TIMESTAMP_NTZ_GROUP + 1, + STRING_GROUP = TIMESTAMP_TZ_GROUP + 1, + GROUP_COUNT = STRING_GROUP + 1, +}; + +struct ScalarGroup { + DataTypePtr type; + MutableColumnPtr values; + DorisVector<size_t> source_rows; +}; + +using ScalarGroups = std::array<ScalarGroup, GROUP_COUNT>; + +template <typename TypeFactory> +ScalarGroup& initialize_group(ScalarGroups& groups, size_t index, TypeFactory&& make_type) { + ScalarGroup& group = groups[index]; + if (!group.values) { + group.type = std::forward<TypeFactory>(make_type)(); + group.values = group.type->create_column(); + } + return group; +} + +void append_invalid(ScalarGroups& groups, size_t row) { + groups[INVALID_GROUP].source_rows.push_back(row); +} + +std::pair<int64_t, uint32_t> split_epoch_micros(int64_t micros) { + constexpr int64_t MICROS_PER_SECOND = 1'000'000; + int64_t seconds = micros / MICROS_PER_SECOND; + int64_t fraction = micros % MICROS_PER_SECOND; + if (fraction < 0) { + --seconds; + fraction += MICROS_PER_SECOND; + } + return {seconds, static_cast<uint32_t>(fraction)}; +} + +int64_t nanos_to_micros(int64_t nanos) { + int64_t micros = nanos / 1000; + if (nanos % 1000 < 0) { + --micros; + } + return micros; +} + +bool epoch_micros_to_civil(int64_t micros, cctz::civil_second* civil, uint32_t* fraction) { + const auto [seconds, micros_fraction] = split_epoch_micros(micros); + const auto lookup = + cctz::utc_time_zone().lookup(cctz::time_point<cctz::seconds>(cctz::seconds(seconds))); + if (lookup.cs.year() < 1 || lookup.cs.year() > 9999) { + return false; + } + *civil = lookup.cs; + *fraction = micros_fraction; + return true; +} + +void append_bool(ScalarGroups& groups, size_t row, bool value) { + auto& group = + initialize_group(groups, BOOL_GROUP, [] { return std::make_shared<DataTypeBool>(); }); + assert_cast<ColumnUInt8&>(*group.values).insert_value(value); + group.source_rows.push_back(row); +} + +void append_int(ScalarGroups& groups, size_t row, int64_t value) { + auto& group = + initialize_group(groups, INT_GROUP, [] { return std::make_shared<DataTypeInt64>(); }); + assert_cast<ColumnInt64&>(*group.values).insert_value(value); + group.source_rows.push_back(row); +} + +void append_float(ScalarGroups& groups, size_t row, float value) { + auto& group = initialize_group(groups, FLOAT_GROUP, + [] { return std::make_shared<DataTypeFloat32>(); }); + assert_cast<ColumnFloat32&>(*group.values).insert_value(value); + group.source_rows.push_back(row); +} + +void append_double(ScalarGroups& groups, size_t row, double value) { + auto& group = initialize_group(groups, DOUBLE_GROUP, + [] { return std::make_shared<DataTypeFloat64>(); }); + assert_cast<ColumnFloat64&>(*group.values).insert_value(value); + group.source_rows.push_back(row); +} + +void append_decimal(ScalarGroups& groups, size_t row, VariantDecimal value) { + if (value.scale >= DECIMAL_SCALE_COUNT) { + append_invalid(groups, row); + return; + } + const size_t index = DECIMAL_GROUP_BEGIN + value.scale; + auto& group = initialize_group(groups, index, [scale = value.scale] { + return std::make_shared<DataTypeDecimal128>(38, scale); + }); + assert_cast<ColumnDecimal128V3&>(*group.values).insert_value(Decimal128V3 {value.unscaled}); + group.source_rows.push_back(row); +} + +void append_date(ScalarGroups& groups, size_t row, int32_t days_since_epoch) { + const cctz::civil_day civil = cctz::civil_day(1970, 1, 1) + days_since_epoch; + if (civil.year() < 1 || civil.year() > 9999) { + append_invalid(groups, row); + return; + } + DateV2Value<DateV2ValueType> value; + value.unchecked_set_time(static_cast<uint16_t>(civil.year()), + static_cast<uint8_t>(civil.month()), static_cast<uint8_t>(civil.day()), + 0, 0, 0); + auto& group = + initialize_group(groups, DATE_GROUP, [] { return std::make_shared<DataTypeDateV2>(); }); + assert_cast<ColumnDateV2&>(*group.values).insert_value(value); + group.source_rows.push_back(row); +} + +void append_timestamp(ScalarGroups& groups, size_t row, int64_t micros, bool utc_adjusted) { + cctz::civil_second civil; + uint32_t fraction = 0; + if (!epoch_micros_to_civil(micros, &civil, &fraction)) { + append_invalid(groups, row); + return; + } + if (utc_adjusted) { + TimestampTzValue value; + value.unchecked_set_time( + static_cast<uint16_t>(civil.year()), static_cast<uint8_t>(civil.month()), + static_cast<uint8_t>(civil.day()), static_cast<uint8_t>(civil.hour()), + static_cast<uint8_t>(civil.minute()), static_cast<uint8_t>(civil.second()), + fraction); + auto& group = initialize_group(groups, TIMESTAMP_TZ_GROUP, + [] { return std::make_shared<DataTypeTimeStampTz>(6); }); + assert_cast<ColumnTimeStampTz&>(*group.values).insert_value(value); + group.source_rows.push_back(row); + return; + } + DateV2Value<DateTimeV2ValueType> value; + value.unchecked_set_time( + static_cast<uint16_t>(civil.year()), static_cast<uint8_t>(civil.month()), + static_cast<uint8_t>(civil.day()), static_cast<uint8_t>(civil.hour()), + static_cast<uint8_t>(civil.minute()), static_cast<uint8_t>(civil.second()), fraction); + auto& group = initialize_group(groups, TIMESTAMP_NTZ_GROUP, + [] { return std::make_shared<DataTypeDateTimeV2>(6); }); + assert_cast<ColumnDateTimeV2&>(*group.values).insert_value(value); + group.source_rows.push_back(row); +} + +void append_string(ScalarGroups& groups, size_t row, StringRef value) { + auto& group = initialize_group(groups, STRING_GROUP, + [] { return std::make_shared<DataTypeString>(); }); + assert_cast<ColumnString&>(*group.values).insert_data(value.data, value.size); + group.source_rows.push_back(row); +} + +void classify_value(ScalarGroups& groups, size_t row, VariantRef value, bool forced_null) { + if (forced_null) { + append_invalid(groups, row); + return; + } + switch (value.basic_type()) { + case VariantBasicType::SHORT_STRING: + append_string(groups, row, value.get_string()); + return; + case VariantBasicType::OBJECT: + case VariantBasicType::ARRAY: + append_invalid(groups, row); + return; + case VariantBasicType::PRIMITIVE: + break; + } + + switch (value.primitive_id()) { + case VariantPrimitiveId::NULL_VALUE: + case VariantPrimitiveId::BINARY: + case VariantPrimitiveId::TIME_NTZ_MICROS: + case VariantPrimitiveId::UUID: + append_invalid(groups, row); + return; + case VariantPrimitiveId::TRUE_VALUE: + append_bool(groups, row, true); + return; + case VariantPrimitiveId::FALSE_VALUE: + append_bool(groups, row, false); + return; + case VariantPrimitiveId::INT8: + case VariantPrimitiveId::INT16: + case VariantPrimitiveId::INT32: + case VariantPrimitiveId::INT64: + append_int(groups, row, value.get_int()); + return; + case VariantPrimitiveId::FLOAT: + append_float(groups, row, value.get_float()); + return; + case VariantPrimitiveId::DOUBLE: + append_double(groups, row, value.get_double()); + return; + case VariantPrimitiveId::DECIMAL4: + case VariantPrimitiveId::DECIMAL8: + case VariantPrimitiveId::DECIMAL16: + append_decimal(groups, row, value.get_decimal()); + return; + case VariantPrimitiveId::DATE: + append_date(groups, row, value.get_date()); + return; + case VariantPrimitiveId::TIMESTAMP_MICROS: + append_timestamp(groups, row, value.get_timestamp_micros(), true); + return; + case VariantPrimitiveId::TIMESTAMP_NTZ_MICROS: + append_timestamp(groups, row, value.get_timestamp_ntz_micros(), false); + return; + case VariantPrimitiveId::TIMESTAMP_NANOS: + append_timestamp(groups, row, nanos_to_micros(value.get_timestamp_nanos()), true); + return; + case VariantPrimitiveId::TIMESTAMP_NTZ_NANOS: + append_timestamp(groups, row, nanos_to_micros(value.get_timestamp_ntz_nanos()), false); + return; + case VariantPrimitiveId::STRING: + append_string(groups, row, value.get_string()); + return; + } + throw Exception(ErrorCode::CORRUPTION, "Unknown Variant primitive id"); +} + +ColumnPtr as_nullable(ColumnPtr column, size_t rows) { + if (check_and_get_column<ColumnNullable>(column.get()) != nullptr) { + return column; + } + return ColumnNullable::create(column, ColumnUInt8::create(rows, 0)); +} + +Status execute_concrete_cast(FunctionContext* context, const ScalarGroup& group, + const DataTypePtr& target_type, ColumnPtr* output) { + if (context == nullptr) { + return Status::InvalidArgument("Variant V2 scalar CAST requires a FunctionContext"); + } + auto cast_context = context->clone(); + cast_context->set_enable_strict_mode(false); + const DataTypePtr nullable_target = make_nullable(target_type); + Block temporary {{group.values->get_ptr(), group.type, "variant-v2-group"}, + {nullable_target->create_column(), nullable_target, "variant-v2-result"}}; + WrapperType wrapper = + prepare_unpack_dictionaries(cast_context.get(), group.type, nullable_target); + Status status = + wrapper(cast_context.get(), temporary, {0}, 1, group.source_rows.size(), nullptr); + if (!status.ok()) { + if (status.is<ErrorCode::INVALID_ARGUMENT>()) { + *output = make_all_null_column(target_type, group.source_rows.size()); + return Status::OK(); + } + return status; + } + *output = as_nullable(temporary.get_by_position(1).column, group.source_rows.size()); + return Status::OK(); +} + +Status assemble_groups(FunctionContext* context, const ScalarGroups& groups, + const DataTypePtr& target_type, size_t rows, ColumnPtr* output) { + const ScalarGroup* only_group = nullptr; + size_t only_group_index = 0; + for (size_t index = 0; index < groups.size(); ++index) { + if (groups[index].source_rows.empty()) { + continue; + } + if (only_group != nullptr) { + only_group = nullptr; + break; + } + only_group = &groups[index]; + only_group_index = index; + } + if (only_group != nullptr && only_group->source_rows.size() == rows) { + if (only_group_index == INVALID_GROUP) { + *output = make_all_null_column(target_type, rows); + return Status::OK(); + } + return execute_concrete_cast(context, *only_group, target_type, output); + } + + MutableColumnPtr nested = target_type->create_column(); + auto nulls = ColumnUInt8::create(); + nested->reserve(rows); + nulls->reserve(rows); + IColumn::Permutation permutation(rows); + size_t concatenated_row = 0; + + for (size_t index = 0; index < groups.size(); ++index) { + const ScalarGroup& group = groups[index]; + if (group.source_rows.empty()) { + continue; + } + ColumnPtr cast_result; + if (index == INVALID_GROUP) { + cast_result = make_all_null_column(target_type, group.source_rows.size()); + } else { + RETURN_IF_ERROR(execute_concrete_cast(context, group, target_type, &cast_result)); + } + const auto& nullable = assert_cast<const ColumnNullable&>(*cast_result); + nested->insert_range_from(nullable.get_nested_column(), 0, group.source_rows.size()); + nulls->insert_range_from(nullable.get_null_map_column(), 0, group.source_rows.size()); + for (size_t source_row : group.source_rows) { + permutation[source_row] = concatenated_row++; + } + } + if (concatenated_row != rows) { + return Status::InternalError("Variant V2 scalar grouping produced {} rows, expected {}", + concatenated_row, rows); + } + ColumnPtr concatenated = ColumnNullable::create(std::move(nested), std::move(nulls)); + *output = concatenated->permute(permutation, rows); + return Status::OK(); +} + +Status execute_typed_cast(FunctionContext* context, const ColumnPtr& source, + const DataTypePtr& source_type, const DataTypePtr& target_type, + size_t rows, ColumnPtr* output) { + if (context == nullptr) { + return Status::InvalidArgument("Variant V2 scalar CAST requires a FunctionContext"); + } + auto cast_context = context->clone(); + cast_context->set_enable_strict_mode(false); + const DataTypePtr nullable_source = make_nullable(source_type); + const DataTypePtr nullable_target = make_nullable(target_type); + Block temporary {{source, nullable_source, "variant-v2-typed"}, + {nullable_target->create_column(), nullable_target, "variant-v2-result"}}; + WrapperType wrapper = + prepare_unpack_dictionaries(cast_context.get(), nullable_source, nullable_target); + Status status = wrapper(cast_context.get(), temporary, {0}, 1, rows, nullptr); + if (!status.ok()) { + if (status.is<ErrorCode::INVALID_ARGUMENT>()) { + *output = make_all_null_column(target_type, rows); + return Status::OK(); + } + return status; + } + *output = as_nullable(temporary.get_by_position(1).column, rows); + return Status::OK(); +} + +} // namespace + +bool is_supported_scalar_source(const DataTypePtr& type) { + switch (remove_nullable(type)->get_primitive_type()) { + case TYPE_BOOLEAN: + case TYPE_TINYINT: + case TYPE_SMALLINT: + case TYPE_INT: + case TYPE_BIGINT: + case TYPE_LARGEINT: + case TYPE_FLOAT: + case TYPE_DOUBLE: + case TYPE_DECIMALV2: + case TYPE_DECIMAL32: + case TYPE_DECIMAL64: + case TYPE_DECIMAL128I: + case TYPE_DATE: + case TYPE_DATEV2: + case TYPE_DATETIME: + case TYPE_DATETIMEV2: + case TYPE_TIMESTAMPTZ: + case TYPE_CHAR: + case TYPE_VARCHAR: + case TYPE_STRING: + return true; + default: + return false; + } +} + +bool is_supported_scalar_target(const DataTypePtr& type) { + switch (remove_nullable(type)->get_primitive_type()) { + case TYPE_BOOLEAN: + case TYPE_TINYINT: + case TYPE_SMALLINT: + case TYPE_INT: + case TYPE_BIGINT: + case TYPE_LARGEINT: + case TYPE_FLOAT: + case TYPE_DOUBLE: + case TYPE_DECIMALV2: + case TYPE_DECIMAL32: + case TYPE_DECIMAL64: + case TYPE_DECIMAL128I: + case TYPE_DECIMAL256: + case TYPE_DATE: + case TYPE_DATEV2: + case TYPE_DATETIME: + case TYPE_DATETIMEV2: + case TYPE_TIMESTAMPTZ: + return true; + default: + return false; + } +} + +Status cast_scalar_to_variant(const ColumnPtr& source, const DataTypePtr& source_type, size_t rows, + ForcedNulls forced_nulls, ColumnPtr* output) { + if (!source || source->size() != rows || + (!forced_nulls.empty() && forced_nulls.size() != rows)) { + return Status::InvalidArgument("Invalid scalar input shape for Variant V2 CAST"); + } + auto nulls = ColumnUInt8::create(rows, 0); + if (!forced_nulls.empty()) { + std::ranges::copy(forced_nulls, nulls->get_data().begin()); + } + ColumnPtr null_map = std::move(nulls); + ColumnPtr nullable = ColumnNullable::create(source, null_map); + *output = ColumnVariantV2::create_typed(std::move(nullable), source_type); + return Status::OK(); +} + +Status cast_typed_variant_to_scalar(FunctionContext* context, const ColumnVariantV2& source, + const DataTypePtr& target_type, size_t rows, + ForcedNulls forced_nulls, ColumnPtr* output) { + if (!source.is_typed() || source.size() != rows) { + return Status::InvalidArgument("Expected a typed Variant V2 source with {} rows", rows); + } + ColumnPtr converted; + RETURN_IF_ERROR(execute_typed_cast(context, source.typed_column().get_ptr(), + source.typed_type(), target_type, rows, &converted)); + return apply_forced_nulls(std::move(converted), forced_nulls, output); +} + +Status cast_variant_refs_to_scalar(FunctionContext* context, std::span<const VariantRef> values, + const DataTypePtr& target_type, ForcedNulls forced_nulls, + ColumnPtr* output) { + if (!forced_nulls.empty() && forced_nulls.size() != values.size()) { + return Status::InvalidArgument("Variant V2 CAST null map has {} rows, expected {}", + forced_nulls.size(), values.size()); + } + ScalarGroups groups; + for (size_t row = 0; row < values.size(); ++row) { + classify_value(groups, row, values[row], !forced_nulls.empty() && forced_nulls[row] != 0); + } + return assemble_groups(context, groups, target_type, values.size(), output); +} + +Status cast_encoded_variant_to_scalar(FunctionContext* context, const ColumnVariantV2& source, + const DataTypePtr& target_type, size_t rows, + ForcedNulls forced_nulls, ColumnPtr* output) { + if (source.is_typed() || source.size() != rows || + (!forced_nulls.empty() && forced_nulls.size() != rows)) { + return Status::InvalidArgument("Invalid encoded Variant V2 input for scalar CAST"); + } + ScalarGroups groups; + for (size_t row = 0; row < rows; ++row) { + classify_value(groups, row, source.get_value_ref(row), + !forced_nulls.empty() && forced_nulls[row] != 0); + } + return assemble_groups(context, groups, target_type, rows, output); +} + +Status cast_variant_values_to_scalar(FunctionContext* context, const ColumnVariantV2& source, + const DataTypePtr& target_type, size_t rows, + ForcedNulls forced_nulls, ColumnPtr* output) { + if (source.size() != rows || (!forced_nulls.empty() && forced_nulls.size() != rows)) { + return Status::InvalidArgument("Invalid Variant V2 input for canonical scalar CAST"); + } + ScalarGroups groups; + column_variant_v2_internal::visit_variant_values( + source, 0, rows, forced_nulls, [&](size_t row) { append_invalid(groups, row); }, + [&](size_t row, VariantRef value) { classify_value(groups, row, value, false); }); + return assemble_groups(context, groups, target_type, rows, output); +} + +ColumnPtr make_all_null_column(const DataTypePtr& nested_type, size_t rows) { + const DataTypePtr concrete_type = remove_nullable(nested_type); + MutableColumnPtr nested = concrete_type->create_column(); + nested->insert_many_defaults(rows); + return ColumnNullable::create(std::move(nested), ColumnUInt8::create(rows, 1)); +} + +Status apply_forced_nulls(ColumnPtr column, ForcedNulls forced_nulls, ColumnPtr* output) { + if (!column) { + return Status::InvalidArgument("Cannot apply a null map to an empty Variant V2 result"); + } + if (forced_nulls.empty()) { + *output = std::move(column); + return Status::OK(); + } + if (forced_nulls.size() != column->size()) { + return Status::InvalidArgument("Variant V2 CAST null map has {} rows, expected {}", + forced_nulls.size(), column->size()); + } + auto nulls = ColumnUInt8::create(column->size(), 0); + if (const auto* nullable = check_and_get_column<ColumnNullable>(column.get())) { + const NullMap& existing = nullable->get_null_map_data(); + for (size_t row = 0; row < column->size(); ++row) { + nulls->get_data()[row] = existing[row] | forced_nulls[row]; + } + *output = ColumnNullable::create(nullable->get_nested_column_ptr(), std::move(nulls)); Review Comment: ColumnNullable::create(column, nullmap) 会自动合并column 的nullamap和第二个参数的nullmap -- 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]
