eldenmoon commented on code in PR #65561: URL: https://github.com/apache/doris/pull/65561#discussion_r3664707550
########## 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) { + 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); Review Comment: This is expected ########## be/src/core/value/variant/variant_canonical.cpp: ########## @@ -0,0 +1,1168 @@ +// 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_canonical.h" + +#include <crc32c/crc32c.h> + +#include <algorithm> +#include <array> +#include <bit> +#include <cmath> +#include <cstdint> +#include <cstring> +#include <limits> +#include <string> +#include <utility> +#include <vector> + +#include "common/exception.h" +#include "core/value/variant/variant_field.h" +#include "core/value/variant/variant_parquet_encoding.h" +#include "exec/common/sip_hash.h" +#include "util/hash_util.hpp" +#include "util/utf8_check.h" + +namespace doris { +namespace { + +constexpr uint64_t CANONICAL_NAN_BITS = 0x7FF8000000000000ULL; + +uint32_t read_bounded_unsigned(StringRef bytes, size_t offset, uint8_t width, const char* field) { + if (offset > bytes.size || width > bytes.size - offset) { + throw Exception(ErrorCode::CORRUPTION, "Truncated Variant canonical cell while reading {}", + field); + } + uint32_t result = 0; + for (uint8_t byte = 0; byte < width; ++byte) { + result |= static_cast<uint32_t>(static_cast<uint8_t>(bytes.data[offset + byte])) + << (byte * 8); + } + return result; +} + +enum class CanonicalKind : uint8_t { + NULL_VALUE = 0, + BOOL = 1, + EXACT_INTEGER = 2, + DECIMAL = 3, + FLOATING = 4, + STRING = 5, + BINARY = 6, + DATE = 7, + TIMESTAMP_TZ = 8, + TIMESTAMP_NTZ = 9, + TIME = 10, + UUID = 11, + OBJECT = 12, + ARRAY = 13, +}; + +struct NormalizedValue { + __int128 integer = 0; + uint64_t floating_bits = 0; + StringRef bytes; + CanonicalKind kind = CanonicalKind::NULL_VALUE; + uint8_t scale = 0; + bool boolean = false; + std::array<uint8_t, 16> uuid {}; +}; + +struct ObjectEntry { + StringRef key; + VariantRef value; +}; + +NormalizedValue normalized_kind(CanonicalKind kind) { + NormalizedValue result; + result.kind = kind; + return result; +} + +NormalizedValue normalized_integer(CanonicalKind kind, __int128 value) { + NormalizedValue result; + result.kind = kind; + result.integer = value; + return result; +} + +NormalizedValue normalized_floating(uint64_t bits) { + NormalizedValue result; + result.kind = CanonicalKind::FLOATING; + result.floating_bits = bits; + return result; +} + +NormalizedValue normalized_bytes(CanonicalKind kind, StringRef bytes) { + NormalizedValue result; + result.kind = kind; + result.bytes = bytes; + return result; +} + +void require_depth(uint32_t depth) { + if (depth > VARIANT_MAX_NESTING_DEPTH) { + throw Exception(ErrorCode::INVALID_ARGUMENT, + "Variant canonical traversal exceeds maximum depth {}", + VARIANT_MAX_NESTING_DEPTH); + } +} + +void require_exact_value(VariantRef value) { + const size_t encoded_size = value.value_size(); + if (encoded_size != value.value.size) { + throw Exception(ErrorCode::CORRUPTION, + "Variant value has {} trailing bytes after its {} byte root", + value.value.size - encoded_size, encoded_size); + } +} + +void require_valid_utf8(StringRef value, const char* description) { + if (value.size != 0 && !validate_utf8(value.data, value.size)) { + throw Exception(ErrorCode::CORRUPTION, "Variant {} is not valid UTF-8", description); + } +} + +NormalizedValue normalize_floating(double value) { + constexpr double INT128_UPPER_EXCLUSIVE = 0x1p127; + if (std::isfinite(value) && std::trunc(value) == value && value >= -INT128_UPPER_EXCLUSIVE && + value < INT128_UPPER_EXCLUSIVE) { + const auto integer = static_cast<__int128>(value); + if (variant_unsigned_magnitude(integer) <= VARIANT_DECIMAL16_MAX) { + return normalized_integer(CanonicalKind::EXACT_INTEGER, integer); + } + } + if (std::isnan(value)) { + return normalized_floating(CANONICAL_NAN_BITS); + } + auto bits = std::bit_cast<uint64_t>(value); + if (value == 0) { + bits = 0; + } + return normalized_floating(bits); +} + +NormalizedValue normalize_primitive(VariantRef value) { + switch (value.primitive_id()) { + case VariantPrimitiveId::NULL_VALUE: + return normalized_kind(CanonicalKind::NULL_VALUE); + case VariantPrimitiveId::TRUE_VALUE: { + NormalizedValue result = normalized_kind(CanonicalKind::BOOL); + result.boolean = true; + return result; + } + case VariantPrimitiveId::FALSE_VALUE: + return normalized_kind(CanonicalKind::BOOL); + case VariantPrimitiveId::INT8: + case VariantPrimitiveId::INT16: + case VariantPrimitiveId::INT32: + case VariantPrimitiveId::INT64: + return normalized_integer(CanonicalKind::EXACT_INTEGER, value.get_int()); + case VariantPrimitiveId::DOUBLE: + return normalize_floating(value.get_double()); + case VariantPrimitiveId::FLOAT: + return normalize_floating(static_cast<double>(value.get_float())); + case VariantPrimitiveId::DECIMAL4: + case VariantPrimitiveId::DECIMAL8: + case VariantPrimitiveId::DECIMAL16: { + VariantDecimal decimal = value.get_decimal(); + if (variant_unsigned_magnitude(decimal.unscaled) > VARIANT_DECIMAL16_MAX) { + throw Exception(ErrorCode::CORRUPTION, + "Variant decimal unscaled value exceeds precision 38"); + } + while (decimal.scale != 0 && decimal.unscaled % 10 == 0) { + decimal.unscaled /= 10; + --decimal.scale; + } + if (decimal.scale == 0) { + return normalized_integer(CanonicalKind::EXACT_INTEGER, decimal.unscaled); + } + NormalizedValue result = normalized_integer(CanonicalKind::DECIMAL, decimal.unscaled); + result.scale = decimal.scale; + return result; + } + case VariantPrimitiveId::DATE: + return normalized_integer(CanonicalKind::DATE, value.get_date()); + case VariantPrimitiveId::TIMESTAMP_MICROS: + return normalized_integer(CanonicalKind::TIMESTAMP_TZ, + static_cast<__int128>(value.get_timestamp_micros()) * 1000); + case VariantPrimitiveId::TIMESTAMP_NTZ_MICROS: + return normalized_integer(CanonicalKind::TIMESTAMP_NTZ, + static_cast<__int128>(value.get_timestamp_ntz_micros()) * 1000); + case VariantPrimitiveId::TIMESTAMP_NANOS: + return normalized_integer(CanonicalKind::TIMESTAMP_TZ, value.get_timestamp_nanos()); + case VariantPrimitiveId::TIMESTAMP_NTZ_NANOS: + return normalized_integer(CanonicalKind::TIMESTAMP_NTZ, value.get_timestamp_ntz_nanos()); + case VariantPrimitiveId::BINARY: + return normalized_bytes(CanonicalKind::BINARY, value.get_binary()); + case VariantPrimitiveId::STRING: { + const StringRef string = value.get_string(); + require_valid_utf8(string, "string"); + return normalized_bytes(CanonicalKind::STRING, string); + } + case VariantPrimitiveId::TIME_NTZ_MICROS: + return normalized_integer(CanonicalKind::TIME, value.get_time_ntz_micros()); + case VariantPrimitiveId::UUID: { + NormalizedValue result = normalized_kind(CanonicalKind::UUID); + result.uuid = value.get_uuid(); + return result; + } + } + throw Exception(ErrorCode::INVALID_ARGUMENT, "Unknown Variant primitive id {}", + static_cast<uint8_t>(value.primitive_id())); +} + +NormalizedValue normalize_value(VariantRef value) { + switch (value.basic_type()) { + case VariantBasicType::SHORT_STRING: { + const StringRef string = value.get_string(); + require_valid_utf8(string, "string"); + return normalized_bytes(CanonicalKind::STRING, string); + } + case VariantBasicType::OBJECT: + return normalized_kind(CanonicalKind::OBJECT); + case VariantBasicType::ARRAY: + return normalized_kind(CanonicalKind::ARRAY); + case VariantBasicType::PRIMITIVE: + return normalize_primitive(value); + } + throw Exception(ErrorCode::INVALID_ARGUMENT, "Unknown Variant basic type"); +} + +ObjectEntry object_entry_at(VariantRef object, uint32_t index, StringRef previous_key, + bool has_previous) { + uint32_t field_id = 0; + VariantRef child = object.object_value_at(index, &field_id); + const StringRef key = object.metadata.key_at(field_id); + require_valid_utf8(key, "object key"); + if (has_previous && previous_key.compare(key) >= 0) { + throw Exception(ErrorCode::CORRUPTION, + "Variant object keys are not strictly byte-sorted at field {}", index); + } + return {.key = key, .value = child}; +} + +bool scalar_equals(const NormalizedValue& left, const NormalizedValue& right) { + switch (left.kind) { + case CanonicalKind::NULL_VALUE: + return true; + case CanonicalKind::BOOL: + return left.boolean == right.boolean; + case CanonicalKind::EXACT_INTEGER: + case CanonicalKind::DATE: + case CanonicalKind::TIMESTAMP_TZ: + case CanonicalKind::TIMESTAMP_NTZ: + case CanonicalKind::TIME: + return left.integer == right.integer; + case CanonicalKind::DECIMAL: + return left.integer == right.integer && left.scale == right.scale; + case CanonicalKind::FLOATING: + return left.floating_bits == right.floating_bits; + case CanonicalKind::STRING: + case CanonicalKind::BINARY: + return left.bytes == right.bytes; + case CanonicalKind::UUID: + return left.uuid == right.uuid; + case CanonicalKind::OBJECT: + case CanonicalKind::ARRAY: + break; + } + DCHECK(false) << "Container reached scalar equality"; + return false; +} + +bool equals_node(VariantRef left, VariantRef right, uint32_t depth) { + require_depth(depth); + require_exact_value(left); + require_exact_value(right); + const NormalizedValue normalized_left = normalize_value(left); + const NormalizedValue normalized_right = normalize_value(right); + if (normalized_left.kind != normalized_right.kind) { + return false; + } + if (normalized_left.kind == CanonicalKind::ARRAY) { + const uint32_t count = left.num_elements(); + if (count != right.num_elements()) { + return false; + } + for (uint32_t index = 0; index < count; ++index) { + if (!equals_node(left.array_at(index), right.array_at(index), depth + 1)) { + return false; + } + } + return true; + } + if (normalized_left.kind == CanonicalKind::OBJECT) { + const uint32_t count = left.num_elements(); + if (count != right.num_elements()) { + return false; + } + StringRef previous_left; + StringRef previous_right; + for (uint32_t index = 0; index < count; ++index) { + const ObjectEntry left_entry = object_entry_at(left, index, previous_left, index != 0); + const ObjectEntry right_entry = + object_entry_at(right, index, previous_right, index != 0); + if (left_entry.key != right_entry.key || + !equals_node(left_entry.value, right_entry.value, depth + 1)) { + return false; + } + previous_left = left_entry.key; + previous_right = right_entry.key; + } + return true; + } + return scalar_equals(normalized_left, normalized_right); +} + +template <typename Sink> +void update_unsigned(Sink& sink, unsigned __int128 value, uint8_t width) { + std::array<char, 16> bytes {}; + for (uint8_t index = 0; index < width; ++index) { + bytes[index] = static_cast<char>(value >> (index * 8)); + } + sink.update(bytes.data(), width); +} + +template <typename Sink> +void update_signed(Sink& sink, __int128 value, uint8_t width) { + update_unsigned(sink, static_cast<unsigned __int128>(value), width); +} + +template <typename Sink> +void update_bytes(Sink& sink, StringRef bytes) { + if (bytes.size > std::numeric_limits<uint32_t>::max()) { + throw Exception(ErrorCode::CORRUPTION, "Variant byte sequence exceeds uint32 length"); + } + update_unsigned(sink, bytes.size, sizeof(uint32_t)); + if (bytes.size != 0) { + sink.update(bytes.data, bytes.size); + } +} + +template <typename Sink> +void hash_normalized_scalar(const NormalizedValue& normalized, Sink& sink) { + switch (normalized.kind) { + case CanonicalKind::NULL_VALUE: + return; + case CanonicalKind::BOOL: { + const char boolean = normalized.boolean ? 1 : 0; + sink.update(&boolean, 1); + return; + } + case CanonicalKind::EXACT_INTEGER: + update_signed(sink, normalized.integer, 16); + return; + case CanonicalKind::DECIMAL: + update_signed(sink, normalized.integer, 16); + sink.update(reinterpret_cast<const char*>(&normalized.scale), 1); + return; + case CanonicalKind::FLOATING: + update_unsigned(sink, normalized.floating_bits, sizeof(uint64_t)); + return; + case CanonicalKind::STRING: + case CanonicalKind::BINARY: + update_bytes(sink, normalized.bytes); + return; + case CanonicalKind::DATE: + update_signed(sink, normalized.integer, sizeof(int32_t)); + return; + case CanonicalKind::TIMESTAMP_TZ: + case CanonicalKind::TIMESTAMP_NTZ: + update_signed(sink, normalized.integer, 16); + return; + case CanonicalKind::TIME: + update_signed(sink, normalized.integer, sizeof(int64_t)); + return; + case CanonicalKind::UUID: + sink.update(reinterpret_cast<const char*>(normalized.uuid.data()), normalized.uuid.size()); + return; + case CanonicalKind::OBJECT: + case CanonicalKind::ARRAY: + break; + } + DCHECK(false) << "Container reached scalar hash"; +} + +template <typename Sink> +void hash_node(VariantRef value, Sink& sink, uint32_t depth) { + require_depth(depth); + require_exact_value(value); + const NormalizedValue normalized = normalize_value(value); + const char tag = static_cast<char>(normalized.kind); + sink.update(&tag, 1); + + switch (normalized.kind) { + case CanonicalKind::OBJECT: { + const uint32_t count = value.num_elements(); + update_unsigned(sink, count, sizeof(uint32_t)); + StringRef previous; + for (uint32_t index = 0; index < count; ++index) { + const ObjectEntry entry = object_entry_at(value, index, previous, index != 0); + update_bytes(sink, entry.key); + hash_node(entry.value, sink, depth + 1); + previous = entry.key; + } + return; + } + case CanonicalKind::ARRAY: { + const uint32_t count = value.num_elements(); + update_unsigned(sink, count, sizeof(uint32_t)); + for (uint32_t index = 0; index < count; ++index) { + hash_node(value.array_at(index), sink, depth + 1); + } + return; + } + default: + hash_normalized_scalar(normalized, sink); + return; + } +} + +uint8_t minimum_integer_width(__int128 value) { + if (value >= std::numeric_limits<int8_t>::min() && + value <= std::numeric_limits<int8_t>::max()) { + return 1; + } + if (value >= std::numeric_limits<int16_t>::min() && + value <= std::numeric_limits<int16_t>::max()) { + return 2; + } + if (value >= std::numeric_limits<int32_t>::min() && + value <= std::numeric_limits<int32_t>::max()) { + return 4; + } + return 8; +} + +uint8_t minimum_decimal_width(__int128 value) { + const unsigned __int128 absolute = variant_unsigned_magnitude(value); + if (absolute <= VARIANT_DECIMAL4_MAX) { + return 4; + } + if (absolute <= VARIANT_DECIMAL8_MAX) { + return 8; + } + return 16; +} + +bool fits_int64(__int128 value) { + return value >= std::numeric_limits<int64_t>::min() && + value <= std::numeric_limits<int64_t>::max(); +} + +size_t scalar_encoded_size(const NormalizedValue& value) { + switch (value.kind) { + case CanonicalKind::NULL_VALUE: + case CanonicalKind::BOOL: + return 1; + case CanonicalKind::EXACT_INTEGER: + return fits_int64(value.integer) ? 1 + minimum_integer_width(value.integer) : 18; + case CanonicalKind::DECIMAL: + return 2 + minimum_decimal_width(value.integer); + case CanonicalKind::FLOATING: + return 1 + sizeof(uint64_t); + case CanonicalKind::STRING: + return value.bytes.size <= VARIANT_MAX_SHORT_STRING_SIZE + ? 1 + value.bytes.size + : 1 + sizeof(uint32_t) + value.bytes.size; + case CanonicalKind::BINARY: + return 1 + sizeof(uint32_t) + value.bytes.size; + case CanonicalKind::DATE: + return 1 + sizeof(int32_t); + case CanonicalKind::TIMESTAMP_TZ: + case CanonicalKind::TIMESTAMP_NTZ: + case CanonicalKind::TIME: + return 1 + sizeof(int64_t); + case CanonicalKind::UUID: + return 1 + 16; + case CanonicalKind::OBJECT: + case CanonicalKind::ARRAY: + break; + } + DCHECK(false) << "Container reached scalar size"; + return 0; +} + +struct PlanNode { + NormalizedValue normalized; + size_t encoded_size = 0; + uint32_t values_size = 0; + size_t children_begin = 0; + uint32_t child_count = 0; + uint8_t count_width = 0; + uint8_t offset_width = 0; + uint8_t id_width = 0; +}; + +struct PlanChild { + uint32_t node_index = 0; + uint32_t field_id = 0; + StringRef key; +}; + +struct SerializePlan { + std::vector<PlanNode> nodes; Review Comment: This is expected -- 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]
