eldenmoon commented on code in PR #65561: URL: https://github.com/apache/doris/pull/65561#discussion_r3644931565
########## 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() { + unsigned __int128 value = 1; + for (uint8_t digit = 0; digit < 38; ++digit) { + value *= 10; + } + return value - 1; +} + +constexpr unsigned __int128 MAX_DECIMAL38 = max_decimal38(); + +uint32_t read_u32(const char* data) noexcept { + uint32_t result = 0; + for (uint8_t byte = 0; byte < METADATA_SIZE_PREFIX; ++byte) { + result |= static_cast<uint32_t>(static_cast<uint8_t>(data[byte])) << (byte * 8); + } + return result; +} + +void write_u32(char* destination, uint32_t value) noexcept { + for (uint8_t byte = 0; byte < METADATA_SIZE_PREFIX; ++byte) { + destination[byte] = static_cast<char>(value >> (byte * 8)); + } +} + +void require_non_null(StringRef bytes, const char* description) { + if (bytes.size != 0 && bytes.data == nullptr) { + throw Exception(ErrorCode::INVALID_ARGUMENT, + "VariantField {} has a null pointer for {} bytes", description, bytes.size); + } +} + +unsigned __int128 magnitude(__int128 value) { + const auto unsigned_value = static_cast<unsigned __int128>(value); + return value < 0 ? ~unsigned_value + 1 : unsigned_value; +} + +void require_valid_utf8(StringRef value, const char* description) { + if (value.size != 0 && !validate_utf8(value.data, value.size)) { + throw Exception(ErrorCode::CORRUPTION, "VariantField {} is not valid UTF-8", description); + } +} + +void require_decimal_in_range(const VariantDecimal& decimal) { + unsigned __int128 maximum = MAX_DECIMAL38; + if (decimal.width == 4) { + maximum = DECIMAL4_MAX; + } else if (decimal.width == 8) { + maximum = DECIMAL8_MAX; + } + if (magnitude(decimal.unscaled) > maximum) { + throw Exception(ErrorCode::CORRUPTION, + "VariantField decimal unscaled value exceeds precision for width {}", + decimal.width); + } +} + +void require_valid_primitive(VariantRef value) { + switch (value.primitive_id()) { + case VariantPrimitiveId::NULL_VALUE: + case VariantPrimitiveId::TRUE_VALUE: + case VariantPrimitiveId::FALSE_VALUE: + return; + case VariantPrimitiveId::INT8: + case VariantPrimitiveId::INT16: + case VariantPrimitiveId::INT32: + case VariantPrimitiveId::INT64: + static_cast<void>(value.get_int()); + return; + case VariantPrimitiveId::DOUBLE: + static_cast<void>(value.get_double()); + return; + case VariantPrimitiveId::DECIMAL4: + case VariantPrimitiveId::DECIMAL8: + case VariantPrimitiveId::DECIMAL16: + require_decimal_in_range(value.get_decimal()); + return; + case VariantPrimitiveId::DATE: + static_cast<void>(value.get_date()); + return; + case VariantPrimitiveId::TIMESTAMP_MICROS: + static_cast<void>(value.get_timestamp_micros()); + return; + case VariantPrimitiveId::TIMESTAMP_NTZ_MICROS: + static_cast<void>(value.get_timestamp_ntz_micros()); + return; + case VariantPrimitiveId::FLOAT: + static_cast<void>(value.get_float()); + return; + case VariantPrimitiveId::BINARY: + static_cast<void>(value.get_binary()); + return; + case VariantPrimitiveId::STRING: + require_valid_utf8(value.get_string(), "string"); + return; + case VariantPrimitiveId::TIME_NTZ_MICROS: + static_cast<void>(value.get_time_ntz_micros()); + return; + case VariantPrimitiveId::TIMESTAMP_NANOS: + static_cast<void>(value.get_timestamp_nanos()); + return; + case VariantPrimitiveId::TIMESTAMP_NTZ_NANOS: + static_cast<void>(value.get_timestamp_ntz_nanos()); + return; + case VariantPrimitiveId::UUID: + static_cast<void>(value.get_uuid()); + return; + } + throw Exception(ErrorCode::CORRUPTION, "VariantField contains an unknown primitive type"); +} + +void require_exact_value(VariantRef value, uint32_t depth); + +struct ObjectValueSpan { + size_t offset; + size_t size; +}; + +size_t object_values_offset(VariantRef value, uint32_t count) { + const uint8_t value_header = + static_cast<uint8_t>(value.value.data[0]) >> VARIANT_VALUE_HEADER_SHIFT; + const auto offset_width = + static_cast<uint8_t>(((value_header >> VARIANT_OBJECT_OFFSET_SIZE_SHIFT) & 0x03U) + 1); + const auto id_width = + static_cast<uint8_t>(((value_header >> VARIANT_OBJECT_ID_SIZE_SHIFT) & 0x03U) + 1); + const size_t count_width = + (value_header & VARIANT_OBJECT_LARGE_MASK) != 0 ? sizeof(uint32_t) : sizeof(uint8_t); + return 1 + count_width + static_cast<size_t>(count) * id_width + + (static_cast<size_t>(count) + 1) * offset_width; +} + +void require_valid_object(VariantRef value, uint32_t depth) { + const uint32_t count = value.num_elements(); + const size_t values_offset = object_values_offset(value, count); + const char* values_begin = value.value.data + values_offset; + const size_t values_size = value.value.size - values_offset; + std::vector<ObjectValueSpan> spans; + spans.reserve(count); + + StringRef previous_key; + for (uint32_t index = 0; index < count; ++index) { + uint32_t field_id = 0; + const VariantRef child = value.object_value_at(index, &field_id); + const StringRef key = value.metadata.key_at(field_id); + if (index != 0 && previous_key.compare(key) >= 0) { + throw Exception(ErrorCode::CORRUPTION, + "VariantField object keys are not strictly ordered at field {}", index); + } + require_exact_value(child, depth + 1); + spans.push_back({.offset = static_cast<size_t>(child.value.data - values_begin), + .size = child.value.size}); + previous_key = key; + } + + std::ranges::sort(spans, [](const ObjectValueSpan& left, const ObjectValueSpan& right) { + return left.offset < right.offset; + }); + size_t consumed = 0; + for (const ObjectValueSpan& span : spans) { + if (span.offset < consumed) { + throw Exception(ErrorCode::CORRUPTION, + "VariantField object value spans overlap or reuse an offset"); + } + if (span.offset > consumed) { + throw Exception(ErrorCode::CORRUPTION, + "VariantField object values contain an unreferenced gap"); + } + consumed += span.size; + } + if (consumed != values_size) { + throw Exception(ErrorCode::CORRUPTION, + "VariantField object values contain unreferenced trailing bytes"); + } +} + +void require_valid_array(VariantRef value, uint32_t depth) { + const uint32_t count = value.num_elements(); + for (uint32_t index = 0; index < count; ++index) { + require_exact_value(value.array_at(index), depth + 1); + } +} + +void require_exact_value(VariantRef value, uint32_t depth) { + if (depth > VARIANT_MAX_NESTING_DEPTH) { + throw Exception(ErrorCode::CORRUPTION, + "VariantField value exceeds maximum nesting depth {}", + VARIANT_MAX_NESTING_DEPTH); + } + const size_t encoded_size = value.value_size(); + if (encoded_size != value.value.size) { + throw Exception(ErrorCode::CORRUPTION, + "VariantField value has {} trailing bytes after its {} byte root", + value.value.size - encoded_size, encoded_size); + } + + switch (value.basic_type()) { + case VariantBasicType::PRIMITIVE: + require_valid_primitive(value); + return; + case VariantBasicType::SHORT_STRING: + require_valid_utf8(value.get_string(), "short string"); + return; + case VariantBasicType::OBJECT: + require_valid_object(value, depth); + return; + case VariantBasicType::ARRAY: + require_valid_array(value, depth); + return; + } + throw Exception(ErrorCode::CORRUPTION, "VariantField contains an unknown value type"); +} + +struct RowSlices { + VariantMetadataRef metadata; + VariantRef value; +}; + +RowSlices split_untrusted(StringRef bytes) { + require_non_null(bytes, "encoded row"); + if (bytes.size < METADATA_SIZE_PREFIX) { + throw Exception(ErrorCode::CORRUPTION, + "Truncated VariantField metadata-size prefix: need {} bytes, have {}", + METADATA_SIZE_PREFIX, bytes.size); + } + const uint32_t metadata_size = read_u32(bytes.data); + const size_t payload_size = bytes.size - METADATA_SIZE_PREFIX; + if (metadata_size > payload_size) { + throw Exception(ErrorCode::CORRUPTION, + "VariantField metadata size {} exceeds {} payload bytes", metadata_size, + payload_size); + } + VariantMetadataRef metadata {.data = bytes.data + METADATA_SIZE_PREFIX, .size = metadata_size}; + VariantRef value {.metadata = metadata, + .value = {metadata.data + metadata.size, payload_size - metadata.size}}; + return {.metadata = metadata, .value = value}; +} + +std::unique_ptr<char[]> copy_bytes(StringRef bytes) { + auto result = std::make_unique<char[]>(bytes.size); + std::memcpy(result.get(), bytes.data, bytes.size); + return result; +} + +[[noreturn]] void throw_comparison_not_supported() { + throw Exception(ErrorCode::NOT_IMPLEMENTED_ERROR, + "Comparison between VariantField values is not supported"); +} + +} // namespace + +void validate_variant_metadata(VariantMetadataRef metadata) { + require_non_null({metadata.data, metadata.size}, "metadata"); + metadata.validate(); + const uint32_t key_count = metadata.dict_size(); + for (uint32_t id = 0; id < key_count; ++id) { + require_valid_utf8(metadata.key_at(id), "metadata key"); + } +} + +void validate_variant_payload(VariantRef value) { + require_non_null(value.value, "value"); + require_exact_value(value, 0); +} + +VariantField::VariantField(std::unique_ptr<char[]> data, size_t size) noexcept + : _data(std::move(data)), _size(size) {} + +VariantField::VariantField(const VariantField& other) : _size(other._size) { + if (_size != 0) { + _data = std::make_unique<char[]>(_size); + std::memcpy(_data.get(), other._data.get(), _size); + } +} + +VariantField::VariantField(VariantField&& other) noexcept + : _data(std::move(other._data)), _size(std::exchange(other._size, 0)) {} + +VariantField& VariantField::operator=(const VariantField& other) { + VariantField copy(other); + swap(copy); + return *this; +} + +VariantField& VariantField::operator=(VariantField&& other) noexcept { + if (this != &other) { + _data = std::move(other._data); + _size = std::exchange(other._size, 0); + } + return *this; +} + +VariantField VariantField::encode(VariantRef value) { Review Comment: 函数名称是否合理， 加上注释 ########## 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() { + unsigned __int128 value = 1; + for (uint8_t digit = 0; digit < 38; ++digit) { + value *= 10; + } + return value - 1; +} + +constexpr unsigned __int128 MAX_DECIMAL38 = max_decimal38(); + +uint32_t read_u32(const char* data) noexcept { + uint32_t result = 0; + for (uint8_t byte = 0; byte < METADATA_SIZE_PREFIX; ++byte) { + result |= static_cast<uint32_t>(static_cast<uint8_t>(data[byte])) << (byte * 8); + } + return result; +} + +void write_u32(char* destination, uint32_t value) noexcept { + for (uint8_t byte = 0; byte < METADATA_SIZE_PREFIX; ++byte) { + destination[byte] = static_cast<char>(value >> (byte * 8)); + } +} + +void require_non_null(StringRef bytes, const char* description) { + if (bytes.size != 0 && bytes.data == nullptr) { + throw Exception(ErrorCode::INVALID_ARGUMENT, + "VariantField {} has a null pointer for {} bytes", description, bytes.size); + } +} + +unsigned __int128 magnitude(__int128 value) { + const auto unsigned_value = static_cast<unsigned __int128>(value); + return value < 0 ? ~unsigned_value + 1 : unsigned_value; +} + +void require_valid_utf8(StringRef value, const char* description) { + if (value.size != 0 && !validate_utf8(value.data, value.size)) { + throw Exception(ErrorCode::CORRUPTION, "VariantField {} is not valid UTF-8", description); + } +} + +void require_decimal_in_range(const VariantDecimal& decimal) { + unsigned __int128 maximum = MAX_DECIMAL38; + if (decimal.width == 4) { + maximum = DECIMAL4_MAX; + } else if (decimal.width == 8) { + maximum = DECIMAL8_MAX; + } + if (magnitude(decimal.unscaled) > maximum) { + throw Exception(ErrorCode::CORRUPTION, + "VariantField decimal unscaled value exceeds precision for width {}", + decimal.width); + } +} + +void require_valid_primitive(VariantRef value) { + switch (value.primitive_id()) { + case VariantPrimitiveId::NULL_VALUE: + case VariantPrimitiveId::TRUE_VALUE: + case VariantPrimitiveId::FALSE_VALUE: + return; + case VariantPrimitiveId::INT8: + case VariantPrimitiveId::INT16: + case VariantPrimitiveId::INT32: + case VariantPrimitiveId::INT64: + static_cast<void>(value.get_int()); + return; + case VariantPrimitiveId::DOUBLE: + static_cast<void>(value.get_double()); + return; + case VariantPrimitiveId::DECIMAL4: + case VariantPrimitiveId::DECIMAL8: + case VariantPrimitiveId::DECIMAL16: + require_decimal_in_range(value.get_decimal()); + return; + case VariantPrimitiveId::DATE: + static_cast<void>(value.get_date()); + return; + case VariantPrimitiveId::TIMESTAMP_MICROS: + static_cast<void>(value.get_timestamp_micros()); + return; + case VariantPrimitiveId::TIMESTAMP_NTZ_MICROS: + static_cast<void>(value.get_timestamp_ntz_micros()); + return; + case VariantPrimitiveId::FLOAT: + static_cast<void>(value.get_float()); + return; + case VariantPrimitiveId::BINARY: + static_cast<void>(value.get_binary()); + return; + case VariantPrimitiveId::STRING: + require_valid_utf8(value.get_string(), "string"); + return; + case VariantPrimitiveId::TIME_NTZ_MICROS: + static_cast<void>(value.get_time_ntz_micros()); + return; + case VariantPrimitiveId::TIMESTAMP_NANOS: + static_cast<void>(value.get_timestamp_nanos()); + return; + case VariantPrimitiveId::TIMESTAMP_NTZ_NANOS: + static_cast<void>(value.get_timestamp_ntz_nanos()); + return; + case VariantPrimitiveId::UUID: + static_cast<void>(value.get_uuid()); + return; + } + throw Exception(ErrorCode::CORRUPTION, "VariantField contains an unknown primitive type"); +} + +void require_exact_value(VariantRef value, uint32_t depth); + +struct ObjectValueSpan { + size_t offset; + size_t size; +}; + +size_t object_values_offset(VariantRef value, uint32_t count) { + const uint8_t value_header = + static_cast<uint8_t>(value.value.data[0]) >> VARIANT_VALUE_HEADER_SHIFT; + const auto offset_width = + static_cast<uint8_t>(((value_header >> VARIANT_OBJECT_OFFSET_SIZE_SHIFT) & 0x03U) + 1); + const auto id_width = + static_cast<uint8_t>(((value_header >> VARIANT_OBJECT_ID_SIZE_SHIFT) & 0x03U) + 1); + const size_t count_width = + (value_header & VARIANT_OBJECT_LARGE_MASK) != 0 ? sizeof(uint32_t) : sizeof(uint8_t); + return 1 + count_width + static_cast<size_t>(count) * id_width + + (static_cast<size_t>(count) + 1) * offset_width; +} + +void require_valid_object(VariantRef value, uint32_t depth) { + const uint32_t count = value.num_elements(); + const size_t values_offset = object_values_offset(value, count); + const char* values_begin = value.value.data + values_offset; + const size_t values_size = value.value.size - values_offset; + std::vector<ObjectValueSpan> spans; + spans.reserve(count); + + StringRef previous_key; + for (uint32_t index = 0; index < count; ++index) { + uint32_t field_id = 0; + const VariantRef child = value.object_value_at(index, &field_id); + const StringRef key = value.metadata.key_at(field_id); + if (index != 0 && previous_key.compare(key) >= 0) { + throw Exception(ErrorCode::CORRUPTION, + "VariantField object keys are not strictly ordered at field {}", index); + } + require_exact_value(child, depth + 1); + spans.push_back({.offset = static_cast<size_t>(child.value.data - values_begin), + .size = child.value.size}); + previous_key = key; + } + + std::ranges::sort(spans, [](const ObjectValueSpan& left, const ObjectValueSpan& right) { + return left.offset < right.offset; + }); + size_t consumed = 0; + for (const ObjectValueSpan& span : spans) { + if (span.offset < consumed) { + throw Exception(ErrorCode::CORRUPTION, + "VariantField object value spans overlap or reuse an offset"); + } + if (span.offset > consumed) { + throw Exception(ErrorCode::CORRUPTION, + "VariantField object values contain an unreferenced gap"); + } + consumed += span.size; + } + if (consumed != values_size) { + throw Exception(ErrorCode::CORRUPTION, + "VariantField object values contain unreferenced trailing bytes"); + } +} + +void require_valid_array(VariantRef value, uint32_t depth) { + const uint32_t count = value.num_elements(); + for (uint32_t index = 0; index < count; ++index) { + require_exact_value(value.array_at(index), depth + 1); + } +} + +void require_exact_value(VariantRef value, uint32_t depth) { + if (depth > VARIANT_MAX_NESTING_DEPTH) { + throw Exception(ErrorCode::CORRUPTION, + "VariantField value exceeds maximum nesting depth {}", + VARIANT_MAX_NESTING_DEPTH); + } + const size_t encoded_size = value.value_size(); + if (encoded_size != value.value.size) { + throw Exception(ErrorCode::CORRUPTION, + "VariantField value has {} trailing bytes after its {} byte root", + value.value.size - encoded_size, encoded_size); + } + + switch (value.basic_type()) { + case VariantBasicType::PRIMITIVE: + require_valid_primitive(value); + return; + case VariantBasicType::SHORT_STRING: + require_valid_utf8(value.get_string(), "short string"); + return; + case VariantBasicType::OBJECT: + require_valid_object(value, depth); + return; + case VariantBasicType::ARRAY: + require_valid_array(value, depth); + return; + } + throw Exception(ErrorCode::CORRUPTION, "VariantField contains an unknown value type"); +} + +struct RowSlices { + VariantMetadataRef metadata; + VariantRef value; +}; + +RowSlices split_untrusted(StringRef bytes) { + require_non_null(bytes, "encoded row"); + if (bytes.size < METADATA_SIZE_PREFIX) { + throw Exception(ErrorCode::CORRUPTION, + "Truncated VariantField metadata-size prefix: need {} bytes, have {}", + METADATA_SIZE_PREFIX, bytes.size); + } + const uint32_t metadata_size = read_u32(bytes.data); + const size_t payload_size = bytes.size - METADATA_SIZE_PREFIX; + if (metadata_size > payload_size) { + throw Exception(ErrorCode::CORRUPTION, + "VariantField metadata size {} exceeds {} payload bytes", metadata_size, + payload_size); + } + VariantMetadataRef metadata {.data = bytes.data + METADATA_SIZE_PREFIX, .size = metadata_size}; + VariantRef value {.metadata = metadata, + .value = {metadata.data + metadata.size, payload_size - metadata.size}}; + return {.metadata = metadata, .value = value}; +} + +std::unique_ptr<char[]> copy_bytes(StringRef bytes) { + auto result = std::make_unique<char[]>(bytes.size); + std::memcpy(result.get(), bytes.data, bytes.size); + return result; +} + +[[noreturn]] void throw_comparison_not_supported() { + throw Exception(ErrorCode::NOT_IMPLEMENTED_ERROR, + "Comparison between VariantField values is not supported"); +} + +} // namespace + +void validate_variant_metadata(VariantMetadataRef metadata) { + require_non_null({metadata.data, metadata.size}, "metadata"); + metadata.validate(); + const uint32_t key_count = metadata.dict_size(); + for (uint32_t id = 0; id < key_count; ++id) { + require_valid_utf8(metadata.key_at(id), "metadata key"); + } +} + +void validate_variant_payload(VariantRef value) { + require_non_null(value.value, "value"); + require_exact_value(value, 0); +} + +VariantField::VariantField(std::unique_ptr<char[]> data, size_t size) noexcept + : _data(std::move(data)), _size(size) {} + +VariantField::VariantField(const VariantField& other) : _size(other._size) { + if (_size != 0) { + _data = std::make_unique<char[]>(_size); + std::memcpy(_data.get(), other._data.get(), _size); + } +} + +VariantField::VariantField(VariantField&& other) noexcept + : _data(std::move(other._data)), _size(std::exchange(other._size, 0)) {} + +VariantField& VariantField::operator=(const VariantField& other) { + VariantField copy(other); + swap(copy); + return *this; +} + +VariantField& VariantField::operator=(VariantField&& other) noexcept { + if (this != &other) { + _data = std::move(other._data); + _size = std::exchange(other._size, 0); + } + return *this; +} + +VariantField VariantField::encode(VariantRef value) { + if (value.metadata.size > std::numeric_limits<uint32_t>::max()) { + throw Exception(ErrorCode::INVALID_ARGUMENT, + "VariantField metadata size {} exceeds uint32 limit", value.metadata.size); + } + if (value.metadata.size > std::numeric_limits<size_t>::max() - METADATA_SIZE_PREFIX) { + throw Exception(ErrorCode::INVALID_ARGUMENT, + "VariantField metadata size exceeds the addressable row size"); + } + const size_t value_offset = METADATA_SIZE_PREFIX + value.metadata.size; + if (value.value.size > std::numeric_limits<size_t>::max() - value_offset) { + throw Exception(ErrorCode::INVALID_ARGUMENT, + "VariantField value size exceeds the addressable row size"); + } + + validate_variant_metadata(value.metadata); + validate_variant_payload(value); + const size_t total_size = value_offset + value.value.size; + auto data = std::make_unique<char[]>(total_size); + write_u32(data.get(), static_cast<uint32_t>(value.metadata.size)); + std::memcpy(data.get() + METADATA_SIZE_PREFIX, value.metadata.data, value.metadata.size); + std::memcpy(data.get() + value_offset, value.value.data, value.value.size); + return {std::move(data), total_size}; +} + +VariantField VariantField::decode(StringRef bytes) { Review Comment: 已将两个入口改名为 `from_ref` / `from_bytes`，并补充了校验、ownership 与不做 canonicalize 的语义注释。 ########## 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: 两者应保持一致。已让 CAST whitelist 直接复用 typed identity whitelist，并补齐 IPv4/IPv6 CAST 用例。 ########## 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_parquet_encoding.h` 的共享 decimal 上限常量。 ########## 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: 已统一到共享 decimal 上限常量，canonical / builder / field / typed path 共用同一份定义。 ########## 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.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: 暂时不用，后面有性能瓶颈再考虑， 大部分情况下 schema是能被去重的 ########## 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() { + unsigned __int128 value = 1; + for (uint8_t digit = 0; digit < 38; ++digit) { + value *= 10; + } + return value - 1; +} + +constexpr unsigned __int128 MAX_DECIMAL38 = max_decimal38(); + +uint32_t read_u32(const char* data) noexcept { + uint32_t result = 0; + for (uint8_t byte = 0; byte < METADATA_SIZE_PREFIX; ++byte) { + result |= static_cast<uint32_t>(static_cast<uint8_t>(data[byte])) << (byte * 8); + } + return result; +} + +void write_u32(char* destination, uint32_t value) noexcept { + for (uint8_t byte = 0; byte < METADATA_SIZE_PREFIX; ++byte) { + destination[byte] = static_cast<char>(value >> (byte * 8)); + } +} + +void require_non_null(StringRef bytes, const char* description) { + if (bytes.size != 0 && bytes.data == nullptr) { + throw Exception(ErrorCode::INVALID_ARGUMENT, + "VariantField {} has a null pointer for {} bytes", description, bytes.size); + } +} + +unsigned __int128 magnitude(__int128 value) { + const auto unsigned_value = static_cast<unsigned __int128>(value); + return value < 0 ? ~unsigned_value + 1 : unsigned_value; +} + +void require_valid_utf8(StringRef value, const char* description) { + if (value.size != 0 && !validate_utf8(value.data, value.size)) { + throw Exception(ErrorCode::CORRUPTION, "VariantField {} is not valid UTF-8", description); + } +} + +void require_decimal_in_range(const VariantDecimal& decimal) { + unsigned __int128 maximum = MAX_DECIMAL38; + if (decimal.width == 4) { + maximum = DECIMAL4_MAX; + } else if (decimal.width == 8) { + maximum = DECIMAL8_MAX; + } + if (magnitude(decimal.unscaled) > maximum) { + throw Exception(ErrorCode::CORRUPTION, + "VariantField decimal unscaled value exceeds precision for width {}", + decimal.width); + } +} + +void require_valid_primitive(VariantRef value) { + switch (value.primitive_id()) { + case VariantPrimitiveId::NULL_VALUE: + case VariantPrimitiveId::TRUE_VALUE: + case VariantPrimitiveId::FALSE_VALUE: + return; + case VariantPrimitiveId::INT8: + case VariantPrimitiveId::INT16: + case VariantPrimitiveId::INT32: + case VariantPrimitiveId::INT64: + static_cast<void>(value.get_int()); + return; + case VariantPrimitiveId::DOUBLE: + static_cast<void>(value.get_double()); + return; + case VariantPrimitiveId::DECIMAL4: + case VariantPrimitiveId::DECIMAL8: + case VariantPrimitiveId::DECIMAL16: + require_decimal_in_range(value.get_decimal()); + return; + case VariantPrimitiveId::DATE: + static_cast<void>(value.get_date()); + return; + case VariantPrimitiveId::TIMESTAMP_MICROS: + static_cast<void>(value.get_timestamp_micros()); + return; + case VariantPrimitiveId::TIMESTAMP_NTZ_MICROS: + static_cast<void>(value.get_timestamp_ntz_micros()); + return; + case VariantPrimitiveId::FLOAT: + static_cast<void>(value.get_float()); + return; + case VariantPrimitiveId::BINARY: + static_cast<void>(value.get_binary()); + return; + case VariantPrimitiveId::STRING: + require_valid_utf8(value.get_string(), "string"); + return; + case VariantPrimitiveId::TIME_NTZ_MICROS: + static_cast<void>(value.get_time_ntz_micros()); + return; + case VariantPrimitiveId::TIMESTAMP_NANOS: + static_cast<void>(value.get_timestamp_nanos()); + return; + case VariantPrimitiveId::TIMESTAMP_NTZ_NANOS: + static_cast<void>(value.get_timestamp_ntz_nanos()); + return; + case VariantPrimitiveId::UUID: + static_cast<void>(value.get_uuid()); + return; + } + throw Exception(ErrorCode::CORRUPTION, "VariantField contains an unknown primitive type"); +} + +void require_exact_value(VariantRef value, uint32_t depth); + +struct ObjectValueSpan { + size_t offset; + size_t size; +}; + +size_t object_values_offset(VariantRef value, uint32_t count) { + const uint8_t value_header = + static_cast<uint8_t>(value.value.data[0]) >> VARIANT_VALUE_HEADER_SHIFT; + const auto offset_width = + static_cast<uint8_t>(((value_header >> VARIANT_OBJECT_OFFSET_SIZE_SHIFT) & 0x03U) + 1); + const auto id_width = + static_cast<uint8_t>(((value_header >> VARIANT_OBJECT_ID_SIZE_SHIFT) & 0x03U) + 1); + const size_t count_width = + (value_header & VARIANT_OBJECT_LARGE_MASK) != 0 ? sizeof(uint32_t) : sizeof(uint8_t); + return 1 + count_width + static_cast<size_t>(count) * id_width + + (static_cast<size_t>(count) + 1) * offset_width; +} + +void require_valid_object(VariantRef value, uint32_t depth) { + const uint32_t count = value.num_elements(); + const size_t values_offset = object_values_offset(value, count); + const char* values_begin = value.value.data + values_offset; + const size_t values_size = value.value.size - values_offset; + std::vector<ObjectValueSpan> spans; + spans.reserve(count); + + StringRef previous_key; + for (uint32_t index = 0; index < count; ++index) { + uint32_t field_id = 0; + const VariantRef child = value.object_value_at(index, &field_id); + const StringRef key = value.metadata.key_at(field_id); + if (index != 0 && previous_key.compare(key) >= 0) { + throw Exception(ErrorCode::CORRUPTION, + "VariantField object keys are not strictly ordered at field {}", index); + } + require_exact_value(child, depth + 1); + spans.push_back({.offset = static_cast<size_t>(child.value.data - values_begin), + .size = child.value.size}); + previous_key = key; + } + + std::ranges::sort(spans, [](const ObjectValueSpan& left, const ObjectValueSpan& right) { + return left.offset < right.offset; + }); + size_t consumed = 0; + for (const ObjectValueSpan& span : spans) { + if (span.offset < consumed) { + throw Exception(ErrorCode::CORRUPTION, + "VariantField object value spans overlap or reuse an offset"); + } + if (span.offset > consumed) { + throw Exception(ErrorCode::CORRUPTION, + "VariantField object values contain an unreferenced gap"); + } + consumed += span.size; + } + if (consumed != values_size) { + throw Exception(ErrorCode::CORRUPTION, + "VariantField object values contain unreferenced trailing bytes"); + } +} + +void require_valid_array(VariantRef value, uint32_t depth) { + const uint32_t count = value.num_elements(); + for (uint32_t index = 0; index < count; ++index) { + require_exact_value(value.array_at(index), depth + 1); + } +} + +void require_exact_value(VariantRef value, uint32_t depth) { + if (depth > VARIANT_MAX_NESTING_DEPTH) { + throw Exception(ErrorCode::CORRUPTION, + "VariantField value exceeds maximum nesting depth {}", + VARIANT_MAX_NESTING_DEPTH); + } + const size_t encoded_size = value.value_size(); + if (encoded_size != value.value.size) { + throw Exception(ErrorCode::CORRUPTION, + "VariantField value has {} trailing bytes after its {} byte root", + value.value.size - encoded_size, encoded_size); + } + + switch (value.basic_type()) { + case VariantBasicType::PRIMITIVE: + require_valid_primitive(value); + return; + case VariantBasicType::SHORT_STRING: + require_valid_utf8(value.get_string(), "short string"); + return; + case VariantBasicType::OBJECT: + require_valid_object(value, depth); + return; + case VariantBasicType::ARRAY: + require_valid_array(value, depth); + return; + } + throw Exception(ErrorCode::CORRUPTION, "VariantField contains an unknown value type"); +} + +struct RowSlices { + VariantMetadataRef metadata; + VariantRef value; +}; + +RowSlices split_untrusted(StringRef bytes) { + require_non_null(bytes, "encoded row"); + if (bytes.size < METADATA_SIZE_PREFIX) { + throw Exception(ErrorCode::CORRUPTION, + "Truncated VariantField metadata-size prefix: need {} bytes, have {}", + METADATA_SIZE_PREFIX, bytes.size); + } + const uint32_t metadata_size = read_u32(bytes.data); + const size_t payload_size = bytes.size - METADATA_SIZE_PREFIX; + if (metadata_size > payload_size) { + throw Exception(ErrorCode::CORRUPTION, + "VariantField metadata size {} exceeds {} payload bytes", metadata_size, + payload_size); + } + VariantMetadataRef metadata {.data = bytes.data + METADATA_SIZE_PREFIX, .size = metadata_size}; + VariantRef value {.metadata = metadata, + .value = {metadata.data + metadata.size, payload_size - metadata.size}}; + return {.metadata = metadata, .value = value}; +} + +std::unique_ptr<char[]> copy_bytes(StringRef bytes) { + auto result = std::make_unique<char[]>(bytes.size); + std::memcpy(result.get(), bytes.data, bytes.size); + return result; +} + +[[noreturn]] void throw_comparison_not_supported() { + throw Exception(ErrorCode::NOT_IMPLEMENTED_ERROR, + "Comparison between VariantField values is not supported"); +} + +} // namespace + +void validate_variant_metadata(VariantMetadataRef metadata) { + require_non_null({metadata.data, metadata.size}, "metadata"); + metadata.validate(); + const uint32_t key_count = metadata.dict_size(); + for (uint32_t id = 0; id < key_count; ++id) { + require_valid_utf8(metadata.key_at(id), "metadata key"); + } +} + +void validate_variant_payload(VariantRef value) { + require_non_null(value.value, "value"); + require_exact_value(value, 0); +} + +VariantField::VariantField(std::unique_ptr<char[]> data, size_t size) noexcept + : _data(std::move(data)), _size(size) {} + +VariantField::VariantField(const VariantField& other) : _size(other._size) { + if (_size != 0) { + _data = std::make_unique<char[]>(_size); + std::memcpy(_data.get(), other._data.get(), _size); + } +} + +VariantField::VariantField(VariantField&& other) noexcept + : _data(std::move(other._data)), _size(std::exchange(other._size, 0)) {} + +VariantField& VariantField::operator=(const VariantField& other) { + VariantField copy(other); + swap(copy); + return *this; +} + +VariantField& VariantField::operator=(VariantField&& other) noexcept { + if (this != &other) { + _data = std::move(other._data); + _size = std::exchange(other._size, 0); + } + return *this; +} + +VariantField VariantField::encode(VariantRef value) { Review Comment: 已将两个入口改名为 `from_ref` / `from_bytes`，并补充了校验、ownership 与不做 canonicalize 的语义注释。 ########## 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: 没有区别，已抽成 Variant V2 CAST 共用的 format-options helper，避免后续漂移。 ########## 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: 第一版是hash map实现，但是比较复杂， 后面来简化 ########## 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: 已统一到 `variant_parquet_encoding.h` 的共享 magnitude helper；JSON 格式化路径中的同类重复也一并清理了。 ########## 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: 语义上会自动合并；但 shared create 会先 clone 传入 null map，再做 OR，多一次 O(N) 分配/拷贝。这里保留单次手工合并，并补了注释和 null-map union 用例。 ########## 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_unsigned`。 -- 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]
