eldenmoon commented on code in PR #65561: URL: https://github.com/apache/doris/pull/65561#discussion_r3664704340
########## be/src/exprs/function/cast/variant_v2/cast_variant_to_scalar.cpp: ########## @@ -0,0 +1,520 @@ +// 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 = DataTypeDecimal128::max_precision() + 1; + +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>(DataTypeDecimal128::max_precision(), 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"); +} + +Status execute_non_strict_scalar_cast(FunctionContext* context, const ColumnPtr& source, + const DataTypePtr& source_type, + const DataTypePtr& target_type, const char* source_name, + 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_target = make_nullable(target_type); + Block temporary {{source, source_type, source_name}, + {nullable_target->create_column(), nullable_target, "variant-v2-result"}}; + WrapperType wrapper = + prepare_unpack_dictionaries(cast_context.get(), source_type, 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 = make_nullable(temporary.get_by_position(1).column); + return Status::OK(); +} + +Status execute_concrete_cast(FunctionContext* context, const ScalarGroup& group, + const DataTypePtr& target_type, ColumnPtr* output) { + return execute_non_strict_scalar_cast(context, group.values->get_ptr(), group.type, target_type, + "variant-v2-group", group.source_rows.size(), output); +} + +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) { + const DataTypePtr nullable_source = make_nullable(source_type); + return execute_non_strict_scalar_cast(context, source, nullable_source, target_type, + "variant-v2-typed", rows, output); +} + +} // namespace + +bool is_supported_scalar_source(const DataTypePtr& type) { + // Every scalar admitted by the typed state has a Parquet Variant mapping in + // with_variant_typed_scalar(), so CAST and the physical typed-state whitelist must stay + // identical. + return is_supported_variant_typed_identity(remove_nullable(type)->get_primitive_type()); +} + +bool is_supported_scalar_target(const DataTypePtr& type) { Review Comment: No such problem ########## be/src/core/column/variant_v2/column_variant_v2_typed_column.h: ########## @@ -0,0 +1,156 @@ +// 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 <cstdint> +#include <limits> +#include <string_view> +#include <utility> + +#include "common/check.h" +#include "common/exception.h" +#include "core/assert_cast.h" +#include "core/call_on_type_index.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/large_int_value.h" +#include "core/value/variant/variant_parquet_encoding.h" +#include "core/value/variant/variant_scalar.h" +#include "exec/common/format_ip.h" + +namespace doris { + +bool is_supported_variant_typed_identity(PrimitiveType type); + +template <typename DateValue> +int32_t variant_days_since_epoch(const DateValue& value, size_t row, std::string_view description) { + if (!value.is_valid_date()) { + throw Exception(ErrorCode::INVALID_ARGUMENT, + "Cannot encode invalid {} value at row {} as Variant", description, row); + } + const int64_t days = value.daynr() - static_cast<int64_t>(calc_daynr(1970, 1, 1)); + DCHECK_GE(days, std::numeric_limits<int32_t>::min()); + DCHECK_LE(days, std::numeric_limits<int32_t>::max()); + return static_cast<int32_t>(days); +} + +template <typename DateTimeValue> +int64_t variant_timestamp_micros(const DateTimeValue& value, size_t row, + std::string_view description) { + constexpr int64_t MICROS_PER_SECOND = 1'000'000; + constexpr int64_t SECONDS_PER_DAY = 86'400; + const int64_t days = variant_days_since_epoch(value, row, description); + const int64_t seconds = + days * SECONDS_PER_DAY + value.hour() * 3600 + value.minute() * 60 + value.second(); + return seconds * MICROS_PER_SECOND + value.microsecond(); +} + +template <PrimitiveType Type, typename Column, typename Callback> +// NOLINTNEXTLINE(readability-function-size) -- centralized compile-time scalar mapping matrix. +void with_variant_typed_scalar(const Column& column, size_t row, uint8_t scale, + Callback&& callback) { + // The callback must consume the scalar synchronously: string refs for LARGEINT/IP textual + // adapters borrow stack storage owned by this function. + if constexpr (Type == TYPE_BOOLEAN) { + const bool value = column.get_data()[row] != 0; + callback(VariantScalarRef::boolean(value)); + } else if constexpr (Type == TYPE_TINYINT || Type == TYPE_SMALLINT || Type == TYPE_INT || + Type == TYPE_BIGINT) { + const auto value = column.get_data()[row]; + callback(VariantScalarRef::integer(value)); + } else if constexpr (Type == TYPE_LARGEINT) { + const __int128 value = column.get_data()[row]; + if (variant_unsigned_magnitude(value) <= VARIANT_DECIMAL16_MAX) { Review Comment: will be solved in Next steps -- 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]
