github-actions[bot] commented on code in PR #65561:
URL: https://github.com/apache/doris/pull/65561#discussion_r3664657993
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/ExplicitlyCastableSignature.java:
##########
@@ -42,6 +44,9 @@ static boolean isExplicitlyCastable(DataType signatureType,
DataType realType) {
/** isExplicitlyCastable */
static boolean isPrimitiveExplicitlyCastable(DataType signatureType,
DataType realType) {
+ if (signatureType instanceof JsonType && realType instanceof
VariantType) {
Review Comment:
[P1] Preserve the legacy Variant-to-JSON binding while V2 is disabled. This
unconditional check also rejects storage-backed V1 arguments, although the
legacy BE Variant-to-JSON cast still exists. The patch has to rewrite existing
`json_extract(v, ...)` and `json_object_flatten(v)` regressions with explicit
casts as a result, so merely merging this default-disabled experiment breaks
established queries in default sessions. Please require an explicit JSON cast
only for `VariantType.isComputeV2()` here and in
`TypeCoercionUtils.implicitCast()` until V1 is removed.
##########
be/src/exprs/table_function/vexplode_v2.cpp:
##########
@@ -49,8 +54,30 @@ VExplodeV2TableFunction::VExplodeV2TableFunction() {
Status VExplodeV2TableFunction::_process_init_variant(Block* block, int
value_column_idx,
int children_column_idx)
{
// explode variant array
- auto column_without_nullable =
remove_nullable(block->get_by_position(value_column_idx).column);
- auto column = column_without_nullable->convert_to_full_column_if_const();
+ auto materialized =
+
block->get_by_position(value_column_idx).column->convert_to_full_column_if_const();
+ std::span<const uint8_t> outer_nulls;
+ const IColumn* nested = materialized.get();
+ if (const auto* nullable = check_and_get_column<ColumnNullable>(nested)) {
+ outer_nulls = nullable->get_null_map_data();
+ nested = &nullable->get_nested_column();
+ }
+ if (const auto* variant_v2 =
check_and_get_column<ColumnVariantV2>(nested)) {
+ auto variant_type = std::make_shared<DataTypeVariantV2>();
+ auto target_type = std::make_shared<DataTypeArray>(variant_type);
+ ColumnPtr array_column;
+
RETURN_IF_ERROR(CastWrapper::variant_v2_internal::cast_variant_to_array(
Review Comment:
[P1] Fix the exhausted-input boundary before sending V2 multi-explode
through this path. For inputs of lengths 1 and 2, `_cur_size` is 2. If this
generator precedes another lateral view, `get_same_many_values()` is used; at
offset 1 its shorter input has `element_size == _cur_offset`, but that method
checks `<` rather than `<=` and then dereferences `nested_nullmap_data[pos]` /
inserts from `pos == nested_column.size()`. This can read the following row or
fail instead of emitting SQL NULL. Please use the same strict `element_size >
_cur_offset` boundary as `get_value()` and add an uneven V2 multi-explode
followed by a second generator.
##########
be/src/exprs/function/function_variant_element.cpp:
##########
@@ -97,10 +100,72 @@ class FunctionVariantElement : public IFunction {
return make_nullable(col);
}
+ // Keep legacy/V2 physical-column dispatch in one entry point so nullable
handling stays shared.
+ // NOLINTNEXTLINE(readability-function-size)
Status execute_impl(FunctionContext* context, Block& block, const
ColumnNumbers& arguments,
uint32_t result, size_t input_rows_count) const
override {
- const auto* variant_col = check_and_get_column<ColumnVariant>(
-
remove_nullable(block.get_by_position(arguments[0]).column).get());
+ const ColumnPtr materialized =
+
block.get_by_position(arguments[0]).column->convert_to_full_column_if_const();
+ const IColumn* physical = materialized.get();
+ std::span<const uint8_t> outer_nulls;
+ if (const auto* nullable =
check_and_get_column<ColumnNullable>(physical)) {
+ outer_nulls = nullable->get_null_map_data();
+ physical = &nullable->get_nested_column();
+ }
+ if (const auto* variant_v2 =
check_and_get_column<ColumnVariantV2>(physical)) {
+ if (block.empty()) {
+ block.replace_by_position(result,
ColumnNullable::create(ColumnVariantV2::create(),
+
ColumnUInt8::create()));
+ return Status::OK();
+ }
+
+ auto replace_with_all_null_result = [&]() {
+ auto null_values = ColumnVariantV2::create();
+ null_values->insert_many_defaults(variant_v2->size());
+ block.replace_by_position(
+ result, ColumnNullable::create(std::move(null_values),
+
ColumnUInt8::create(variant_v2->size(), 1)));
+ };
+ const auto& index_argument = block.get_by_position(arguments[1]);
+ const ColumnPtr materialized_index =
+ index_argument.column->convert_to_full_column_if_const();
+ const IColumn* index_column = materialized_index.get();
+ if (index_column->is_null_at(0)) {
+ replace_with_all_null_result();
+ return Status::OK();
+ }
+ if (const auto* nullable =
check_and_get_column<ColumnNullable>(*index_column)) {
+ index_column = &nullable->get_nested_column();
+ }
+
+ std::optional<VariantElementV2PathSegment> segment;
+ const PrimitiveType index_type =
+ remove_nullable(index_argument.type)->get_primitive_type();
+ if (is_string_type(index_type)) {
+ segment =
VariantElementV2PathSegment::object_key(index_column->get_data_at(0));
Review Comment:
[P1] Align FE with the constant-selector contract for Variant. Nereids only
requires a literal second argument for structs, so
`element_at(parse_to_variant('[10,20]'), number + 1)` resolves the new
Variant/BigInt signature. BE still declares argument 1 always constant and
rejects the query before this code with `Argument at index 1 ... must be
constant`. This branch also reads selector row 0 only, so simply dropping that
BE guard would be incorrect. Either reject nonconstant Variant selectors during
analysis or resolve/extract per row, and add a multi-row selector regression.
##########
be/src/core/value/variant/variant_canonical.cpp:
##########
@@ -0,0 +1,1168 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "core/value/variant/variant_canonical.h"
+
+#include <crc32c/crc32c.h>
+
+#include <algorithm>
+#include <array>
+#include <bit>
+#include <cmath>
+#include <cstdint>
+#include <cstring>
+#include <limits>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "common/exception.h"
+#include "core/value/variant/variant_field.h"
+#include "core/value/variant/variant_parquet_encoding.h"
+#include "exec/common/sip_hash.h"
+#include "util/hash_util.hpp"
+#include "util/utf8_check.h"
+
+namespace doris {
+namespace {
+
+constexpr uint64_t CANONICAL_NAN_BITS = 0x7FF8000000000000ULL;
+
+uint32_t read_bounded_unsigned(StringRef bytes, size_t offset, uint8_t width,
const char* field) {
+ if (offset > bytes.size || width > bytes.size - offset) {
+ throw Exception(ErrorCode::CORRUPTION, "Truncated Variant canonical
cell while reading {}",
+ field);
+ }
+ uint32_t result = 0;
+ for (uint8_t byte = 0; byte < width; ++byte) {
+ result |= static_cast<uint32_t>(static_cast<uint8_t>(bytes.data[offset
+ byte]))
+ << (byte * 8);
+ }
+ return result;
+}
+
+enum class CanonicalKind : uint8_t {
+ NULL_VALUE = 0,
+ BOOL = 1,
+ EXACT_INTEGER = 2,
+ DECIMAL = 3,
+ FLOATING = 4,
+ STRING = 5,
+ BINARY = 6,
+ DATE = 7,
+ TIMESTAMP_TZ = 8,
+ TIMESTAMP_NTZ = 9,
+ TIME = 10,
+ UUID = 11,
+ OBJECT = 12,
+ ARRAY = 13,
+};
+
+struct NormalizedValue {
+ __int128 integer = 0;
+ uint64_t floating_bits = 0;
+ StringRef bytes;
+ CanonicalKind kind = CanonicalKind::NULL_VALUE;
+ uint8_t scale = 0;
+ bool boolean = false;
+ std::array<uint8_t, 16> uuid {};
+};
+
+struct ObjectEntry {
+ StringRef key;
+ VariantRef value;
+};
+
+NormalizedValue normalized_kind(CanonicalKind kind) {
+ NormalizedValue result;
+ result.kind = kind;
+ return result;
+}
+
+NormalizedValue normalized_integer(CanonicalKind kind, __int128 value) {
+ NormalizedValue result;
+ result.kind = kind;
+ result.integer = value;
+ return result;
+}
+
+NormalizedValue normalized_floating(uint64_t bits) {
+ NormalizedValue result;
+ result.kind = CanonicalKind::FLOATING;
+ result.floating_bits = bits;
+ return result;
+}
+
+NormalizedValue normalized_bytes(CanonicalKind kind, StringRef bytes) {
+ NormalizedValue result;
+ result.kind = kind;
+ result.bytes = bytes;
+ return result;
+}
+
+void require_depth(uint32_t depth) {
+ if (depth > VARIANT_MAX_NESTING_DEPTH) {
+ throw Exception(ErrorCode::INVALID_ARGUMENT,
+ "Variant canonical traversal exceeds maximum depth {}",
+ VARIANT_MAX_NESTING_DEPTH);
+ }
+}
+
+void require_exact_value(VariantRef value) {
+ const size_t encoded_size = value.value_size();
+ if (encoded_size != value.value.size) {
+ throw Exception(ErrorCode::CORRUPTION,
+ "Variant value has {} trailing bytes after its {} byte
root",
+ value.value.size - encoded_size, encoded_size);
+ }
+}
+
+void require_valid_utf8(StringRef value, const char* description) {
+ if (value.size != 0 && !validate_utf8(value.data, value.size)) {
+ throw Exception(ErrorCode::CORRUPTION, "Variant {} is not valid
UTF-8", description);
+ }
+}
+
+NormalizedValue normalize_floating(double value) {
+ constexpr double INT128_UPPER_EXCLUSIVE = 0x1p127;
+ if (std::isfinite(value) && std::trunc(value) == value && value >=
-INT128_UPPER_EXCLUSIVE &&
+ value < INT128_UPPER_EXCLUSIVE) {
+ const auto integer = static_cast<__int128>(value);
+ if (variant_unsigned_magnitude(integer) <= VARIANT_DECIMAL16_MAX) {
+ return normalized_integer(CanonicalKind::EXACT_INTEGER, integer);
+ }
+ }
+ if (std::isnan(value)) {
+ return normalized_floating(CANONICAL_NAN_BITS);
+ }
+ auto bits = std::bit_cast<uint64_t>(value);
+ if (value == 0) {
+ bits = 0;
+ }
+ return normalized_floating(bits);
+}
+
+NormalizedValue normalize_primitive(VariantRef value) {
+ switch (value.primitive_id()) {
+ case VariantPrimitiveId::NULL_VALUE:
+ return normalized_kind(CanonicalKind::NULL_VALUE);
+ case VariantPrimitiveId::TRUE_VALUE: {
+ NormalizedValue result = normalized_kind(CanonicalKind::BOOL);
+ result.boolean = true;
+ return result;
+ }
+ case VariantPrimitiveId::FALSE_VALUE:
+ return normalized_kind(CanonicalKind::BOOL);
+ case VariantPrimitiveId::INT8:
+ case VariantPrimitiveId::INT16:
+ case VariantPrimitiveId::INT32:
+ case VariantPrimitiveId::INT64:
+ return normalized_integer(CanonicalKind::EXACT_INTEGER,
value.get_int());
+ case VariantPrimitiveId::DOUBLE:
+ return normalize_floating(value.get_double());
+ case VariantPrimitiveId::FLOAT:
+ return normalize_floating(static_cast<double>(value.get_float()));
+ case VariantPrimitiveId::DECIMAL4:
+ case VariantPrimitiveId::DECIMAL8:
+ case VariantPrimitiveId::DECIMAL16: {
+ VariantDecimal decimal = value.get_decimal();
+ if (variant_unsigned_magnitude(decimal.unscaled) >
VARIANT_DECIMAL16_MAX) {
+ throw Exception(ErrorCode::CORRUPTION,
+ "Variant decimal unscaled value exceeds precision
38");
+ }
+ while (decimal.scale != 0 && decimal.unscaled % 10 == 0) {
+ decimal.unscaled /= 10;
+ --decimal.scale;
+ }
+ if (decimal.scale == 0) {
+ return normalized_integer(CanonicalKind::EXACT_INTEGER,
decimal.unscaled);
+ }
+ NormalizedValue result = normalized_integer(CanonicalKind::DECIMAL,
decimal.unscaled);
+ result.scale = decimal.scale;
+ return result;
+ }
+ case VariantPrimitiveId::DATE:
+ return normalized_integer(CanonicalKind::DATE, value.get_date());
+ case VariantPrimitiveId::TIMESTAMP_MICROS:
+ return normalized_integer(CanonicalKind::TIMESTAMP_TZ,
+
static_cast<__int128>(value.get_timestamp_micros()) * 1000);
+ case VariantPrimitiveId::TIMESTAMP_NTZ_MICROS:
+ return normalized_integer(CanonicalKind::TIMESTAMP_NTZ,
+
static_cast<__int128>(value.get_timestamp_ntz_micros()) * 1000);
+ case VariantPrimitiveId::TIMESTAMP_NANOS:
+ return normalized_integer(CanonicalKind::TIMESTAMP_TZ,
value.get_timestamp_nanos());
+ case VariantPrimitiveId::TIMESTAMP_NTZ_NANOS:
+ return normalized_integer(CanonicalKind::TIMESTAMP_NTZ,
value.get_timestamp_ntz_nanos());
+ case VariantPrimitiveId::BINARY:
+ return normalized_bytes(CanonicalKind::BINARY, value.get_binary());
+ case VariantPrimitiveId::STRING: {
+ const StringRef string = value.get_string();
+ require_valid_utf8(string, "string");
+ return normalized_bytes(CanonicalKind::STRING, string);
+ }
+ case VariantPrimitiveId::TIME_NTZ_MICROS:
+ return normalized_integer(CanonicalKind::TIME,
value.get_time_ntz_micros());
+ case VariantPrimitiveId::UUID: {
+ NormalizedValue result = normalized_kind(CanonicalKind::UUID);
+ result.uuid = value.get_uuid();
+ return result;
+ }
+ }
+ throw Exception(ErrorCode::INVALID_ARGUMENT, "Unknown Variant primitive id
{}",
+ static_cast<uint8_t>(value.primitive_id()));
+}
+
+NormalizedValue normalize_value(VariantRef value) {
+ switch (value.basic_type()) {
+ case VariantBasicType::SHORT_STRING: {
+ const StringRef string = value.get_string();
+ require_valid_utf8(string, "string");
+ return normalized_bytes(CanonicalKind::STRING, string);
+ }
+ case VariantBasicType::OBJECT:
+ return normalized_kind(CanonicalKind::OBJECT);
+ case VariantBasicType::ARRAY:
+ return normalized_kind(CanonicalKind::ARRAY);
+ case VariantBasicType::PRIMITIVE:
+ return normalize_primitive(value);
+ }
+ throw Exception(ErrorCode::INVALID_ARGUMENT, "Unknown Variant basic type");
+}
+
+ObjectEntry object_entry_at(VariantRef object, uint32_t index, StringRef
previous_key,
+ bool has_previous) {
+ uint32_t field_id = 0;
+ VariantRef child = object.object_value_at(index, &field_id);
+ const StringRef key = object.metadata.key_at(field_id);
+ require_valid_utf8(key, "object key");
+ if (has_previous && previous_key.compare(key) >= 0) {
+ throw Exception(ErrorCode::CORRUPTION,
+ "Variant object keys are not strictly byte-sorted at
field {}", index);
+ }
+ return {.key = key, .value = child};
+}
+
+bool scalar_equals(const NormalizedValue& left, const NormalizedValue& right) {
+ switch (left.kind) {
+ case CanonicalKind::NULL_VALUE:
+ return true;
+ case CanonicalKind::BOOL:
+ return left.boolean == right.boolean;
+ case CanonicalKind::EXACT_INTEGER:
+ case CanonicalKind::DATE:
+ case CanonicalKind::TIMESTAMP_TZ:
+ case CanonicalKind::TIMESTAMP_NTZ:
+ case CanonicalKind::TIME:
+ return left.integer == right.integer;
+ case CanonicalKind::DECIMAL:
+ return left.integer == right.integer && left.scale == right.scale;
+ case CanonicalKind::FLOATING:
+ return left.floating_bits == right.floating_bits;
+ case CanonicalKind::STRING:
+ case CanonicalKind::BINARY:
+ return left.bytes == right.bytes;
+ case CanonicalKind::UUID:
+ return left.uuid == right.uuid;
+ case CanonicalKind::OBJECT:
+ case CanonicalKind::ARRAY:
+ break;
+ }
+ DCHECK(false) << "Container reached scalar equality";
+ return false;
+}
+
+bool equals_node(VariantRef left, VariantRef right, uint32_t depth) {
+ require_depth(depth);
+ require_exact_value(left);
+ require_exact_value(right);
+ const NormalizedValue normalized_left = normalize_value(left);
+ const NormalizedValue normalized_right = normalize_value(right);
+ if (normalized_left.kind != normalized_right.kind) {
+ return false;
+ }
+ if (normalized_left.kind == CanonicalKind::ARRAY) {
+ const uint32_t count = left.num_elements();
+ if (count != right.num_elements()) {
+ return false;
+ }
+ for (uint32_t index = 0; index < count; ++index) {
+ if (!equals_node(left.array_at(index), right.array_at(index),
depth + 1)) {
+ return false;
+ }
+ }
+ return true;
+ }
+ if (normalized_left.kind == CanonicalKind::OBJECT) {
+ const uint32_t count = left.num_elements();
+ if (count != right.num_elements()) {
+ return false;
+ }
+ StringRef previous_left;
+ StringRef previous_right;
+ for (uint32_t index = 0; index < count; ++index) {
+ const ObjectEntry left_entry = object_entry_at(left, index,
previous_left, index != 0);
+ const ObjectEntry right_entry =
+ object_entry_at(right, index, previous_right, index != 0);
+ if (left_entry.key != right_entry.key ||
+ !equals_node(left_entry.value, right_entry.value, depth + 1)) {
+ return false;
+ }
+ previous_left = left_entry.key;
+ previous_right = right_entry.key;
+ }
+ return true;
+ }
+ return scalar_equals(normalized_left, normalized_right);
+}
+
+template <typename Sink>
+void update_unsigned(Sink& sink, unsigned __int128 value, uint8_t width) {
+ std::array<char, 16> bytes {};
+ for (uint8_t index = 0; index < width; ++index) {
+ bytes[index] = static_cast<char>(value >> (index * 8));
+ }
+ sink.update(bytes.data(), width);
+}
+
+template <typename Sink>
+void update_signed(Sink& sink, __int128 value, uint8_t width) {
+ update_unsigned(sink, static_cast<unsigned __int128>(value), width);
+}
+
+template <typename Sink>
+void update_bytes(Sink& sink, StringRef bytes) {
+ if (bytes.size > std::numeric_limits<uint32_t>::max()) {
+ throw Exception(ErrorCode::CORRUPTION, "Variant byte sequence exceeds
uint32 length");
+ }
+ update_unsigned(sink, bytes.size, sizeof(uint32_t));
+ if (bytes.size != 0) {
+ sink.update(bytes.data, bytes.size);
+ }
+}
+
+template <typename Sink>
+void hash_normalized_scalar(const NormalizedValue& normalized, Sink& sink) {
+ switch (normalized.kind) {
+ case CanonicalKind::NULL_VALUE:
+ return;
+ case CanonicalKind::BOOL: {
+ const char boolean = normalized.boolean ? 1 : 0;
+ sink.update(&boolean, 1);
+ return;
+ }
+ case CanonicalKind::EXACT_INTEGER:
+ update_signed(sink, normalized.integer, 16);
+ return;
+ case CanonicalKind::DECIMAL:
+ update_signed(sink, normalized.integer, 16);
+ sink.update(reinterpret_cast<const char*>(&normalized.scale), 1);
+ return;
+ case CanonicalKind::FLOATING:
+ update_unsigned(sink, normalized.floating_bits, sizeof(uint64_t));
+ return;
+ case CanonicalKind::STRING:
+ case CanonicalKind::BINARY:
+ update_bytes(sink, normalized.bytes);
+ return;
+ case CanonicalKind::DATE:
+ update_signed(sink, normalized.integer, sizeof(int32_t));
+ return;
+ case CanonicalKind::TIMESTAMP_TZ:
+ case CanonicalKind::TIMESTAMP_NTZ:
+ update_signed(sink, normalized.integer, 16);
+ return;
+ case CanonicalKind::TIME:
+ update_signed(sink, normalized.integer, sizeof(int64_t));
+ return;
+ case CanonicalKind::UUID:
+ sink.update(reinterpret_cast<const char*>(normalized.uuid.data()),
normalized.uuid.size());
+ return;
+ case CanonicalKind::OBJECT:
+ case CanonicalKind::ARRAY:
+ break;
+ }
+ DCHECK(false) << "Container reached scalar hash";
+}
+
+template <typename Sink>
+void hash_node(VariantRef value, Sink& sink, uint32_t depth) {
+ require_depth(depth);
+ require_exact_value(value);
+ const NormalizedValue normalized = normalize_value(value);
+ const char tag = static_cast<char>(normalized.kind);
+ sink.update(&tag, 1);
+
+ switch (normalized.kind) {
+ case CanonicalKind::OBJECT: {
+ const uint32_t count = value.num_elements();
+ update_unsigned(sink, count, sizeof(uint32_t));
+ StringRef previous;
+ for (uint32_t index = 0; index < count; ++index) {
+ const ObjectEntry entry = object_entry_at(value, index, previous,
index != 0);
+ update_bytes(sink, entry.key);
+ hash_node(entry.value, sink, depth + 1);
+ previous = entry.key;
+ }
+ return;
+ }
+ case CanonicalKind::ARRAY: {
+ const uint32_t count = value.num_elements();
+ update_unsigned(sink, count, sizeof(uint32_t));
+ for (uint32_t index = 0; index < count; ++index) {
+ hash_node(value.array_at(index), sink, depth + 1);
+ }
+ return;
+ }
+ default:
+ hash_normalized_scalar(normalized, sink);
+ return;
+ }
+}
+
+uint8_t minimum_integer_width(__int128 value) {
+ if (value >= std::numeric_limits<int8_t>::min() &&
+ value <= std::numeric_limits<int8_t>::max()) {
+ return 1;
+ }
+ if (value >= std::numeric_limits<int16_t>::min() &&
+ value <= std::numeric_limits<int16_t>::max()) {
+ return 2;
+ }
+ if (value >= std::numeric_limits<int32_t>::min() &&
+ value <= std::numeric_limits<int32_t>::max()) {
+ return 4;
+ }
+ return 8;
+}
+
+uint8_t minimum_decimal_width(__int128 value) {
+ const unsigned __int128 absolute = variant_unsigned_magnitude(value);
+ if (absolute <= VARIANT_DECIMAL4_MAX) {
+ return 4;
+ }
+ if (absolute <= VARIANT_DECIMAL8_MAX) {
+ return 8;
+ }
+ return 16;
+}
+
+bool fits_int64(__int128 value) {
+ return value >= std::numeric_limits<int64_t>::min() &&
+ value <= std::numeric_limits<int64_t>::max();
+}
+
+size_t scalar_encoded_size(const NormalizedValue& value) {
+ switch (value.kind) {
+ case CanonicalKind::NULL_VALUE:
+ case CanonicalKind::BOOL:
+ return 1;
+ case CanonicalKind::EXACT_INTEGER:
+ return fits_int64(value.integer) ? 1 +
minimum_integer_width(value.integer) : 18;
+ case CanonicalKind::DECIMAL:
+ return 2 + minimum_decimal_width(value.integer);
+ case CanonicalKind::FLOATING:
+ return 1 + sizeof(uint64_t);
+ case CanonicalKind::STRING:
+ return value.bytes.size <= VARIANT_MAX_SHORT_STRING_SIZE
+ ? 1 + value.bytes.size
+ : 1 + sizeof(uint32_t) + value.bytes.size;
+ case CanonicalKind::BINARY:
+ return 1 + sizeof(uint32_t) + value.bytes.size;
+ case CanonicalKind::DATE:
+ return 1 + sizeof(int32_t);
+ case CanonicalKind::TIMESTAMP_TZ:
+ case CanonicalKind::TIMESTAMP_NTZ:
+ case CanonicalKind::TIME:
+ return 1 + sizeof(int64_t);
+ case CanonicalKind::UUID:
+ return 1 + 16;
+ case CanonicalKind::OBJECT:
+ case CanonicalKind::ARRAY:
+ break;
+ }
+ DCHECK(false) << "Container reached scalar size";
+ return 0;
+}
+
+struct PlanNode {
+ NormalizedValue normalized;
+ size_t encoded_size = 0;
+ uint32_t values_size = 0;
+ size_t children_begin = 0;
+ uint32_t child_count = 0;
+ uint8_t count_width = 0;
+ uint8_t offset_width = 0;
+ uint8_t id_width = 0;
+};
+
+struct PlanChild {
+ uint32_t node_index = 0;
+ uint32_t field_id = 0;
+ StringRef key;
+};
+
+struct SerializePlan {
+ std::vector<PlanNode> nodes;
Review Comment:
[P2] Avoid rebuilding heap-backed canonical plans twice for every encoded
Variant key. The serialized GROUP BY/DISTINCT setup first calls
`get_max_row_byte_size()`, which invokes this planner for every row, and then
`serialize()` invokes it again for every row. Each call creates a fresh `Impl`
plus three `std::vector`s; even scalar encoded keys allocate, while wide
objects rebuild O(nodes + children + keys) scratch on both passes. Please reuse
a planner workspace or combine sizing/writing (with a scalar fast path where
possible), and add a representative encoded-key benchmark.
##########
be/src/exprs/function/cast/variant_v2/cast_variant_to_scalar.cpp:
##########
@@ -0,0 +1,520 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include <cctz/time_zone.h>
+
+#include <algorithm>
+#include <array>
+#include <cstdint>
+#include <memory>
+#include <utility>
+
+#include "common/exception.h"
+#include "core/assert_cast.h"
+#include "core/block/block.h"
+#include "core/column/column_decimal.h"
+#include "core/column/column_string.h"
+#include "core/column/column_vector.h"
+#include "core/column/variant_v2/column_variant_v2.h"
+#include "core/custom_allocator.h"
+#include "core/data_type/data_type_date_or_datetime_v2.h"
+#include "core/data_type/data_type_decimal.h"
+#include "core/data_type/data_type_nullable.h"
+#include "core/data_type/data_type_number.h"
+#include "core/data_type/data_type_string.h"
+#include "core/data_type/data_type_timestamptz.h"
+#include "core/value/timestamptz_value.h"
+#include "core/value/vdatetime_value.h"
+#include "exprs/function/cast/cast_base.h"
+#include "exprs/function/cast/variant_v2/cast_variant_v2_internal.h"
+#include "exprs/function_context.h"
+
+namespace doris::CastWrapper::variant_v2_internal {
+namespace {
+
+constexpr size_t DECIMAL_SCALE_COUNT = DataTypeDecimal128::max_precision() + 1;
+
+enum GroupIndex : size_t {
+ INVALID_GROUP = 0,
+ BOOL_GROUP = INVALID_GROUP + 1,
+ INT_GROUP = BOOL_GROUP + 1,
+ FLOAT_GROUP = INT_GROUP + 1,
+ DOUBLE_GROUP = FLOAT_GROUP + 1,
+ DECIMAL_GROUP_BEGIN = DOUBLE_GROUP + 1,
+ DATE_GROUP = DECIMAL_GROUP_BEGIN + DECIMAL_SCALE_COUNT,
+ TIMESTAMP_NTZ_GROUP = DATE_GROUP + 1,
+ TIMESTAMP_TZ_GROUP = TIMESTAMP_NTZ_GROUP + 1,
+ STRING_GROUP = TIMESTAMP_TZ_GROUP + 1,
+ GROUP_COUNT = STRING_GROUP + 1,
+};
+
+struct ScalarGroup {
+ DataTypePtr type;
+ MutableColumnPtr values;
+ DorisVector<size_t> source_rows;
+};
+
+using ScalarGroups = std::array<ScalarGroup, GROUP_COUNT>;
+
+template <typename TypeFactory>
+ScalarGroup& initialize_group(ScalarGroups& groups, size_t index,
TypeFactory&& make_type) {
+ ScalarGroup& group = groups[index];
+ if (!group.values) {
+ group.type = std::forward<TypeFactory>(make_type)();
+ group.values = group.type->create_column();
+ }
+ return group;
+}
+
+void append_invalid(ScalarGroups& groups, size_t row) {
+ groups[INVALID_GROUP].source_rows.push_back(row);
+}
+
+std::pair<int64_t, uint32_t> split_epoch_micros(int64_t micros) {
+ constexpr int64_t MICROS_PER_SECOND = 1'000'000;
+ int64_t seconds = micros / MICROS_PER_SECOND;
+ int64_t fraction = micros % MICROS_PER_SECOND;
+ if (fraction < 0) {
+ --seconds;
+ fraction += MICROS_PER_SECOND;
+ }
+ return {seconds, static_cast<uint32_t>(fraction)};
+}
+
+int64_t nanos_to_micros(int64_t nanos) {
+ int64_t micros = nanos / 1000;
+ if (nanos % 1000 < 0) {
+ --micros;
+ }
+ return micros;
+}
+
+bool epoch_micros_to_civil(int64_t micros, cctz::civil_second* civil,
uint32_t* fraction) {
+ const auto [seconds, micros_fraction] = split_epoch_micros(micros);
+ const auto lookup =
+
cctz::utc_time_zone().lookup(cctz::time_point<cctz::seconds>(cctz::seconds(seconds)));
+ if (lookup.cs.year() < 1 || lookup.cs.year() > 9999) {
+ return false;
+ }
+ *civil = lookup.cs;
+ *fraction = micros_fraction;
+ return true;
+}
+
+void append_bool(ScalarGroups& groups, size_t row, bool value) {
+ auto& group =
+ initialize_group(groups, BOOL_GROUP, [] { return
std::make_shared<DataTypeBool>(); });
+ assert_cast<ColumnUInt8&>(*group.values).insert_value(value);
+ group.source_rows.push_back(row);
+}
+
+void append_int(ScalarGroups& groups, size_t row, int64_t value) {
+ auto& group =
+ initialize_group(groups, INT_GROUP, [] { return
std::make_shared<DataTypeInt64>(); });
+ assert_cast<ColumnInt64&>(*group.values).insert_value(value);
+ group.source_rows.push_back(row);
+}
+
+void append_float(ScalarGroups& groups, size_t row, float value) {
+ auto& group = initialize_group(groups, FLOAT_GROUP,
+ [] { return
std::make_shared<DataTypeFloat32>(); });
+ assert_cast<ColumnFloat32&>(*group.values).insert_value(value);
+ group.source_rows.push_back(row);
+}
+
+void append_double(ScalarGroups& groups, size_t row, double value) {
+ auto& group = initialize_group(groups, DOUBLE_GROUP,
+ [] { return
std::make_shared<DataTypeFloat64>(); });
+ assert_cast<ColumnFloat64&>(*group.values).insert_value(value);
+ group.source_rows.push_back(row);
+}
+
+void append_decimal(ScalarGroups& groups, size_t row, VariantDecimal value) {
+ if (value.scale >= DECIMAL_SCALE_COUNT) {
+ append_invalid(groups, row);
+ return;
+ }
+ const size_t index = DECIMAL_GROUP_BEGIN + value.scale;
+ auto& group = initialize_group(groups, index, [scale = value.scale] {
+ return
std::make_shared<DataTypeDecimal128>(DataTypeDecimal128::max_precision(),
scale);
+ });
+ assert_cast<ColumnDecimal128V3&>(*group.values).insert_value(Decimal128V3
{value.unscaled});
+ group.source_rows.push_back(row);
+}
+
+void append_date(ScalarGroups& groups, size_t row, int32_t days_since_epoch) {
+ const cctz::civil_day civil = cctz::civil_day(1970, 1, 1) +
days_since_epoch;
+ if (civil.year() < 1 || civil.year() > 9999) {
+ append_invalid(groups, row);
+ return;
+ }
+ DateV2Value<DateV2ValueType> value;
+ value.unchecked_set_time(static_cast<uint16_t>(civil.year()),
+ static_cast<uint8_t>(civil.month()),
static_cast<uint8_t>(civil.day()),
+ 0, 0, 0);
+ auto& group =
+ initialize_group(groups, DATE_GROUP, [] { return
std::make_shared<DataTypeDateV2>(); });
+ assert_cast<ColumnDateV2&>(*group.values).insert_value(value);
+ group.source_rows.push_back(row);
+}
+
+void append_timestamp(ScalarGroups& groups, size_t row, int64_t micros, bool
utc_adjusted) {
+ cctz::civil_second civil;
+ uint32_t fraction = 0;
+ if (!epoch_micros_to_civil(micros, &civil, &fraction)) {
+ append_invalid(groups, row);
+ return;
+ }
+ if (utc_adjusted) {
+ TimestampTzValue value;
+ value.unchecked_set_time(
+ static_cast<uint16_t>(civil.year()),
static_cast<uint8_t>(civil.month()),
+ static_cast<uint8_t>(civil.day()),
static_cast<uint8_t>(civil.hour()),
+ static_cast<uint8_t>(civil.minute()),
static_cast<uint8_t>(civil.second()),
+ fraction);
+ auto& group = initialize_group(groups, TIMESTAMP_TZ_GROUP,
+ [] { return
std::make_shared<DataTypeTimeStampTz>(6); });
+ assert_cast<ColumnTimeStampTz&>(*group.values).insert_value(value);
+ group.source_rows.push_back(row);
+ return;
+ }
+ DateV2Value<DateTimeV2ValueType> value;
+ value.unchecked_set_time(
+ static_cast<uint16_t>(civil.year()),
static_cast<uint8_t>(civil.month()),
+ static_cast<uint8_t>(civil.day()),
static_cast<uint8_t>(civil.hour()),
+ static_cast<uint8_t>(civil.minute()),
static_cast<uint8_t>(civil.second()), fraction);
+ auto& group = initialize_group(groups, TIMESTAMP_NTZ_GROUP,
+ [] { return
std::make_shared<DataTypeDateTimeV2>(6); });
+ assert_cast<ColumnDateTimeV2&>(*group.values).insert_value(value);
+ group.source_rows.push_back(row);
+}
+
+void append_string(ScalarGroups& groups, size_t row, StringRef value) {
+ auto& group = initialize_group(groups, STRING_GROUP,
+ [] { return
std::make_shared<DataTypeString>(); });
+ assert_cast<ColumnString&>(*group.values).insert_data(value.data,
value.size);
+ group.source_rows.push_back(row);
+}
+
+void classify_value(ScalarGroups& groups, size_t row, VariantRef value, bool
forced_null) {
+ if (forced_null) {
+ append_invalid(groups, row);
+ return;
+ }
+ switch (value.basic_type()) {
+ case VariantBasicType::SHORT_STRING:
+ append_string(groups, row, value.get_string());
+ return;
+ case VariantBasicType::OBJECT:
+ case VariantBasicType::ARRAY:
+ append_invalid(groups, row);
+ return;
+ case VariantBasicType::PRIMITIVE:
+ break;
+ }
+
+ switch (value.primitive_id()) {
+ case VariantPrimitiveId::NULL_VALUE:
+ case VariantPrimitiveId::BINARY:
+ case VariantPrimitiveId::TIME_NTZ_MICROS:
+ case VariantPrimitiveId::UUID:
+ append_invalid(groups, row);
+ return;
+ case VariantPrimitiveId::TRUE_VALUE:
+ append_bool(groups, row, true);
+ return;
+ case VariantPrimitiveId::FALSE_VALUE:
+ append_bool(groups, row, false);
+ return;
+ case VariantPrimitiveId::INT8:
+ case VariantPrimitiveId::INT16:
+ case VariantPrimitiveId::INT32:
+ case VariantPrimitiveId::INT64:
+ append_int(groups, row, value.get_int());
+ return;
+ case VariantPrimitiveId::FLOAT:
+ append_float(groups, row, value.get_float());
+ return;
+ case VariantPrimitiveId::DOUBLE:
+ append_double(groups, row, value.get_double());
+ return;
+ case VariantPrimitiveId::DECIMAL4:
+ case VariantPrimitiveId::DECIMAL8:
+ case VariantPrimitiveId::DECIMAL16:
+ append_decimal(groups, row, value.get_decimal());
+ return;
+ case VariantPrimitiveId::DATE:
+ append_date(groups, row, value.get_date());
+ return;
+ case VariantPrimitiveId::TIMESTAMP_MICROS:
+ append_timestamp(groups, row, value.get_timestamp_micros(), true);
+ return;
+ case VariantPrimitiveId::TIMESTAMP_NTZ_MICROS:
+ append_timestamp(groups, row, value.get_timestamp_ntz_micros(), false);
+ return;
+ case VariantPrimitiveId::TIMESTAMP_NANOS:
+ append_timestamp(groups, row,
nanos_to_micros(value.get_timestamp_nanos()), true);
+ return;
+ case VariantPrimitiveId::TIMESTAMP_NTZ_NANOS:
+ append_timestamp(groups, row,
nanos_to_micros(value.get_timestamp_ntz_nanos()), false);
+ return;
+ case VariantPrimitiveId::STRING:
+ append_string(groups, row, value.get_string());
+ return;
+ }
+ throw Exception(ErrorCode::CORRUPTION, "Unknown Variant primitive id");
+}
+
+Status execute_non_strict_scalar_cast(FunctionContext* context, const
ColumnPtr& source,
+ const DataTypePtr& source_type,
+ const DataTypePtr& target_type, const
char* source_name,
+ size_t rows, ColumnPtr* output) {
+ if (context == nullptr) {
+ return Status::InvalidArgument("Variant V2 scalar CAST requires a
FunctionContext");
+ }
+ auto cast_context = context->clone();
+ cast_context->set_enable_strict_mode(false);
+ const DataTypePtr nullable_target = make_nullable(target_type);
+ Block temporary {{source, source_type, source_name},
+ {nullable_target->create_column(), nullable_target,
"variant-v2-result"}};
+ WrapperType wrapper =
+ prepare_unpack_dictionaries(cast_context.get(), source_type,
nullable_target);
+ Status status = wrapper(cast_context.get(), temporary, {0}, 1, rows,
nullptr);
+ if (!status.ok()) {
+ if (status.is<ErrorCode::INVALID_ARGUMENT>()) {
+ *output = make_all_null_column(target_type, rows);
+ return Status::OK();
+ }
+ return status;
+ }
+ *output = make_nullable(temporary.get_by_position(1).column);
+ return Status::OK();
+}
+
+Status execute_concrete_cast(FunctionContext* context, const ScalarGroup&
group,
+ const DataTypePtr& target_type, ColumnPtr*
output) {
+ return execute_non_strict_scalar_cast(context, group.values->get_ptr(),
group.type, target_type,
+ "variant-v2-group",
group.source_rows.size(), output);
+}
+
+Status assemble_groups(FunctionContext* context, const ScalarGroups& groups,
+ const DataTypePtr& target_type, size_t rows, ColumnPtr*
output) {
+ const ScalarGroup* only_group = nullptr;
+ size_t only_group_index = 0;
+ for (size_t index = 0; index < groups.size(); ++index) {
+ if (groups[index].source_rows.empty()) {
+ continue;
+ }
+ if (only_group != nullptr) {
+ only_group = nullptr;
+ break;
+ }
+ only_group = &groups[index];
+ only_group_index = index;
+ }
+ if (only_group != nullptr && only_group->source_rows.size() == rows) {
+ if (only_group_index == INVALID_GROUP) {
+ *output = make_all_null_column(target_type, rows);
+ return Status::OK();
+ }
+ return execute_concrete_cast(context, *only_group, target_type,
output);
+ }
+
+ MutableColumnPtr nested = target_type->create_column();
+ auto nulls = ColumnUInt8::create();
+ nested->reserve(rows);
+ nulls->reserve(rows);
+ IColumn::Permutation permutation(rows);
+ size_t concatenated_row = 0;
+
+ for (size_t index = 0; index < groups.size(); ++index) {
+ const ScalarGroup& group = groups[index];
+ if (group.source_rows.empty()) {
+ continue;
+ }
+ ColumnPtr cast_result;
+ if (index == INVALID_GROUP) {
+ cast_result = make_all_null_column(target_type,
group.source_rows.size());
+ } else {
+ RETURN_IF_ERROR(execute_concrete_cast(context, group, target_type,
&cast_result));
+ }
+ const auto& nullable = assert_cast<const
ColumnNullable&>(*cast_result);
+ nested->insert_range_from(nullable.get_nested_column(), 0,
group.source_rows.size());
+ nulls->insert_range_from(nullable.get_null_map_column(), 0,
group.source_rows.size());
+ for (size_t source_row : group.source_rows) {
+ permutation[source_row] = concatenated_row++;
+ }
+ }
+ if (concatenated_row != rows) {
+ return Status::InternalError("Variant V2 scalar grouping produced {}
rows, expected {}",
+ concatenated_row, rows);
+ }
+ ColumnPtr concatenated = ColumnNullable::create(std::move(nested),
std::move(nulls));
+ *output = concatenated->permute(permutation, rows);
+ return Status::OK();
+}
+
+Status execute_typed_cast(FunctionContext* context, const ColumnPtr& source,
+ const DataTypePtr& source_type, const DataTypePtr&
target_type,
+ size_t rows, ColumnPtr* output) {
+ const DataTypePtr nullable_source = make_nullable(source_type);
+ return execute_non_strict_scalar_cast(context, source, nullable_source,
target_type,
+ "variant-v2-typed", rows, output);
+}
+
+} // namespace
+
+bool is_supported_scalar_source(const DataTypePtr& type) {
+ // Every scalar admitted by the typed state has a Parquet Variant mapping
in
+ // with_variant_typed_scalar(), so CAST and the physical typed-state
whitelist must stay
+ // identical.
+ return
is_supported_variant_typed_identity(remove_nullable(type)->get_primitive_type());
+}
+
+bool is_supported_scalar_target(const DataTypePtr& type) {
Review Comment:
[P1] Make FE enforce this narrower V2 cast matrix. Nereids currently accepts
Variant to every primitive/array and also accepts sources such as DECIMAL256,
TIMEV2, MAP, and STRUCT for `CAST(... AS VARIANT)`. These V2 kernels omit
TIME/IP/VARBINARY targets, omit DECIMAL256/TIME sources, and have no MAP/STRUCT
source branch, so those expressions analyze and then return `Conversion ... is
not supported` from BE (the new DECIMAL256 test even expects that runtime
error). Please define the supported matrix once on both sides—either implement
the missing directions or reject them during analysis—and add paired
FE/execution boundary tests.
##########
be/src/core/column/variant_v2/column_variant_v2_typed_column.h:
##########
@@ -0,0 +1,156 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#pragma once
+
+#include <array>
+#include <cstdint>
+#include <limits>
+#include <string_view>
+#include <utility>
+
+#include "common/check.h"
+#include "common/exception.h"
+#include "core/assert_cast.h"
+#include "core/call_on_type_index.h"
+#include "core/column/column_decimal.h"
+#include "core/column/column_nullable.h"
+#include "core/column/column_string.h"
+#include "core/column/column_vector.h"
+#include "core/data_type/data_type.h"
+#include "core/value/large_int_value.h"
+#include "core/value/variant/variant_parquet_encoding.h"
+#include "core/value/variant/variant_scalar.h"
+#include "exec/common/format_ip.h"
+
+namespace doris {
+
+bool is_supported_variant_typed_identity(PrimitiveType type);
+
+template <typename DateValue>
+int32_t variant_days_since_epoch(const DateValue& value, size_t row,
std::string_view description) {
+ if (!value.is_valid_date()) {
+ throw Exception(ErrorCode::INVALID_ARGUMENT,
+ "Cannot encode invalid {} value at row {} as Variant",
description, row);
+ }
+ const int64_t days = value.daynr() - static_cast<int64_t>(calc_daynr(1970,
1, 1));
+ DCHECK_GE(days, std::numeric_limits<int32_t>::min());
+ DCHECK_LE(days, std::numeric_limits<int32_t>::max());
+ return static_cast<int32_t>(days);
+}
+
+template <typename DateTimeValue>
+int64_t variant_timestamp_micros(const DateTimeValue& value, size_t row,
+ std::string_view description) {
+ constexpr int64_t MICROS_PER_SECOND = 1'000'000;
+ constexpr int64_t SECONDS_PER_DAY = 86'400;
+ const int64_t days = variant_days_since_epoch(value, row, description);
+ const int64_t seconds =
+ days * SECONDS_PER_DAY + value.hour() * 3600 + value.minute() * 60
+ value.second();
+ return seconds * MICROS_PER_SECOND + value.microsecond();
+}
+
+template <PrimitiveType Type, typename Column, typename Callback>
+// NOLINTNEXTLINE(readability-function-size) -- centralized compile-time
scalar mapping matrix.
+void with_variant_typed_scalar(const Column& column, size_t row, uint8_t scale,
+ Callback&& callback) {
+ // The callback must consume the scalar synchronously: string refs for
LARGEINT/IP textual
+ // adapters borrow stack storage owned by this function.
+ if constexpr (Type == TYPE_BOOLEAN) {
+ const bool value = column.get_data()[row] != 0;
+ callback(VariantScalarRef::boolean(value));
+ } else if constexpr (Type == TYPE_TINYINT || Type == TYPE_SMALLINT || Type
== TYPE_INT ||
+ Type == TYPE_BIGINT) {
+ const auto value = column.get_data()[row];
+ callback(VariantScalarRef::integer(value));
+ } else if constexpr (Type == TYPE_LARGEINT) {
+ const __int128 value = column.get_data()[row];
+ if (variant_unsigned_magnitude(value) <= VARIANT_DECIMAL16_MAX) {
Review Comment:
[P1] Do not encode an out-of-range numeric LARGEINT as a Variant STRING. At
`10^38`, this branch makes `CAST(CAST('100000000000000000000000000000000000000'
AS LARGEINT) AS VARIANT)` canonically identical to
`CAST(CAST('100000000000000000000000000000000000000' AS STRING) AS VARIANT)`,
because both hashing and arena serialization consume the resulting
`VariantScalarRef::string`. GROUP BY, DISTINCT, and multi-distinct can
therefore collapse a numeric value with text, while the value at `10^38-1`
remains correctly distinct. Please reject the unsupported numeric magnitude or
preserve a numeric representation, and add E/T equality plus GROUP BY/DISTINCT
coverage against the same digit string.
##########
be/src/exprs/function/cast/variant_v2/cast_variant_to_scalar.cpp:
##########
@@ -0,0 +1,520 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include <cctz/time_zone.h>
+
+#include <algorithm>
+#include <array>
+#include <cstdint>
+#include <memory>
+#include <utility>
+
+#include "common/exception.h"
+#include "core/assert_cast.h"
+#include "core/block/block.h"
+#include "core/column/column_decimal.h"
+#include "core/column/column_string.h"
+#include "core/column/column_vector.h"
+#include "core/column/variant_v2/column_variant_v2.h"
+#include "core/custom_allocator.h"
+#include "core/data_type/data_type_date_or_datetime_v2.h"
+#include "core/data_type/data_type_decimal.h"
+#include "core/data_type/data_type_nullable.h"
+#include "core/data_type/data_type_number.h"
+#include "core/data_type/data_type_string.h"
+#include "core/data_type/data_type_timestamptz.h"
+#include "core/value/timestamptz_value.h"
+#include "core/value/vdatetime_value.h"
+#include "exprs/function/cast/cast_base.h"
+#include "exprs/function/cast/variant_v2/cast_variant_v2_internal.h"
+#include "exprs/function_context.h"
+
+namespace doris::CastWrapper::variant_v2_internal {
+namespace {
+
+constexpr size_t DECIMAL_SCALE_COUNT = DataTypeDecimal128::max_precision() + 1;
+
+enum GroupIndex : size_t {
+ INVALID_GROUP = 0,
+ BOOL_GROUP = INVALID_GROUP + 1,
+ INT_GROUP = BOOL_GROUP + 1,
+ FLOAT_GROUP = INT_GROUP + 1,
+ DOUBLE_GROUP = FLOAT_GROUP + 1,
+ DECIMAL_GROUP_BEGIN = DOUBLE_GROUP + 1,
+ DATE_GROUP = DECIMAL_GROUP_BEGIN + DECIMAL_SCALE_COUNT,
+ TIMESTAMP_NTZ_GROUP = DATE_GROUP + 1,
+ TIMESTAMP_TZ_GROUP = TIMESTAMP_NTZ_GROUP + 1,
+ STRING_GROUP = TIMESTAMP_TZ_GROUP + 1,
+ GROUP_COUNT = STRING_GROUP + 1,
+};
+
+struct ScalarGroup {
+ DataTypePtr type;
+ MutableColumnPtr values;
+ DorisVector<size_t> source_rows;
+};
+
+using ScalarGroups = std::array<ScalarGroup, GROUP_COUNT>;
+
+template <typename TypeFactory>
+ScalarGroup& initialize_group(ScalarGroups& groups, size_t index,
TypeFactory&& make_type) {
+ ScalarGroup& group = groups[index];
+ if (!group.values) {
+ group.type = std::forward<TypeFactory>(make_type)();
+ group.values = group.type->create_column();
+ }
+ return group;
+}
+
+void append_invalid(ScalarGroups& groups, size_t row) {
+ groups[INVALID_GROUP].source_rows.push_back(row);
+}
+
+std::pair<int64_t, uint32_t> split_epoch_micros(int64_t micros) {
+ constexpr int64_t MICROS_PER_SECOND = 1'000'000;
+ int64_t seconds = micros / MICROS_PER_SECOND;
+ int64_t fraction = micros % MICROS_PER_SECOND;
+ if (fraction < 0) {
+ --seconds;
+ fraction += MICROS_PER_SECOND;
+ }
+ return {seconds, static_cast<uint32_t>(fraction)};
+}
+
+int64_t nanos_to_micros(int64_t nanos) {
+ int64_t micros = nanos / 1000;
+ if (nanos % 1000 < 0) {
+ --micros;
+ }
+ return micros;
+}
+
+bool epoch_micros_to_civil(int64_t micros, cctz::civil_second* civil,
uint32_t* fraction) {
+ const auto [seconds, micros_fraction] = split_epoch_micros(micros);
+ const auto lookup =
+
cctz::utc_time_zone().lookup(cctz::time_point<cctz::seconds>(cctz::seconds(seconds)));
+ if (lookup.cs.year() < 1 || lookup.cs.year() > 9999) {
+ return false;
+ }
+ *civil = lookup.cs;
+ *fraction = micros_fraction;
+ return true;
+}
+
+void append_bool(ScalarGroups& groups, size_t row, bool value) {
+ auto& group =
+ initialize_group(groups, BOOL_GROUP, [] { return
std::make_shared<DataTypeBool>(); });
+ assert_cast<ColumnUInt8&>(*group.values).insert_value(value);
+ group.source_rows.push_back(row);
+}
+
+void append_int(ScalarGroups& groups, size_t row, int64_t value) {
+ auto& group =
+ initialize_group(groups, INT_GROUP, [] { return
std::make_shared<DataTypeInt64>(); });
+ assert_cast<ColumnInt64&>(*group.values).insert_value(value);
+ group.source_rows.push_back(row);
+}
+
+void append_float(ScalarGroups& groups, size_t row, float value) {
+ auto& group = initialize_group(groups, FLOAT_GROUP,
+ [] { return
std::make_shared<DataTypeFloat32>(); });
+ assert_cast<ColumnFloat32&>(*group.values).insert_value(value);
+ group.source_rows.push_back(row);
+}
+
+void append_double(ScalarGroups& groups, size_t row, double value) {
+ auto& group = initialize_group(groups, DOUBLE_GROUP,
+ [] { return
std::make_shared<DataTypeFloat64>(); });
+ assert_cast<ColumnFloat64&>(*group.values).insert_value(value);
+ group.source_rows.push_back(row);
+}
+
+void append_decimal(ScalarGroups& groups, size_t row, VariantDecimal value) {
+ if (value.scale >= DECIMAL_SCALE_COUNT) {
+ append_invalid(groups, row);
+ return;
+ }
+ const size_t index = DECIMAL_GROUP_BEGIN + value.scale;
+ auto& group = initialize_group(groups, index, [scale = value.scale] {
+ return
std::make_shared<DataTypeDecimal128>(DataTypeDecimal128::max_precision(),
scale);
+ });
+ assert_cast<ColumnDecimal128V3&>(*group.values).insert_value(Decimal128V3
{value.unscaled});
+ group.source_rows.push_back(row);
+}
+
+void append_date(ScalarGroups& groups, size_t row, int32_t days_since_epoch) {
+ const cctz::civil_day civil = cctz::civil_day(1970, 1, 1) +
days_since_epoch;
+ if (civil.year() < 1 || civil.year() > 9999) {
+ append_invalid(groups, row);
+ return;
+ }
+ DateV2Value<DateV2ValueType> value;
+ value.unchecked_set_time(static_cast<uint16_t>(civil.year()),
+ static_cast<uint8_t>(civil.month()),
static_cast<uint8_t>(civil.day()),
+ 0, 0, 0);
+ auto& group =
+ initialize_group(groups, DATE_GROUP, [] { return
std::make_shared<DataTypeDateV2>(); });
+ assert_cast<ColumnDateV2&>(*group.values).insert_value(value);
+ group.source_rows.push_back(row);
+}
+
+void append_timestamp(ScalarGroups& groups, size_t row, int64_t micros, bool
utc_adjusted) {
+ cctz::civil_second civil;
+ uint32_t fraction = 0;
+ if (!epoch_micros_to_civil(micros, &civil, &fraction)) {
+ append_invalid(groups, row);
+ return;
+ }
+ if (utc_adjusted) {
+ TimestampTzValue value;
+ value.unchecked_set_time(
+ static_cast<uint16_t>(civil.year()),
static_cast<uint8_t>(civil.month()),
+ static_cast<uint8_t>(civil.day()),
static_cast<uint8_t>(civil.hour()),
+ static_cast<uint8_t>(civil.minute()),
static_cast<uint8_t>(civil.second()),
+ fraction);
+ auto& group = initialize_group(groups, TIMESTAMP_TZ_GROUP,
+ [] { return
std::make_shared<DataTypeTimeStampTz>(6); });
+ assert_cast<ColumnTimeStampTz&>(*group.values).insert_value(value);
+ group.source_rows.push_back(row);
+ return;
+ }
+ DateV2Value<DateTimeV2ValueType> value;
+ value.unchecked_set_time(
+ static_cast<uint16_t>(civil.year()),
static_cast<uint8_t>(civil.month()),
+ static_cast<uint8_t>(civil.day()),
static_cast<uint8_t>(civil.hour()),
+ static_cast<uint8_t>(civil.minute()),
static_cast<uint8_t>(civil.second()), fraction);
+ auto& group = initialize_group(groups, TIMESTAMP_NTZ_GROUP,
+ [] { return
std::make_shared<DataTypeDateTimeV2>(6); });
+ assert_cast<ColumnDateTimeV2&>(*group.values).insert_value(value);
+ group.source_rows.push_back(row);
+}
+
+void append_string(ScalarGroups& groups, size_t row, StringRef value) {
+ auto& group = initialize_group(groups, STRING_GROUP,
+ [] { return
std::make_shared<DataTypeString>(); });
+ assert_cast<ColumnString&>(*group.values).insert_data(value.data,
value.size);
+ group.source_rows.push_back(row);
+}
+
+void classify_value(ScalarGroups& groups, size_t row, VariantRef value, bool
forced_null) {
+ if (forced_null) {
+ append_invalid(groups, row);
+ return;
+ }
+ switch (value.basic_type()) {
+ case VariantBasicType::SHORT_STRING:
+ append_string(groups, row, value.get_string());
+ return;
+ case VariantBasicType::OBJECT:
+ case VariantBasicType::ARRAY:
+ append_invalid(groups, row);
+ return;
+ case VariantBasicType::PRIMITIVE:
+ break;
+ }
+
+ switch (value.primitive_id()) {
+ case VariantPrimitiveId::NULL_VALUE:
+ case VariantPrimitiveId::BINARY:
+ case VariantPrimitiveId::TIME_NTZ_MICROS:
+ case VariantPrimitiveId::UUID:
+ append_invalid(groups, row);
+ return;
+ case VariantPrimitiveId::TRUE_VALUE:
+ append_bool(groups, row, true);
+ return;
+ case VariantPrimitiveId::FALSE_VALUE:
+ append_bool(groups, row, false);
+ return;
+ case VariantPrimitiveId::INT8:
+ case VariantPrimitiveId::INT16:
+ case VariantPrimitiveId::INT32:
+ case VariantPrimitiveId::INT64:
+ append_int(groups, row, value.get_int());
+ return;
+ case VariantPrimitiveId::FLOAT:
+ append_float(groups, row, value.get_float());
+ return;
+ case VariantPrimitiveId::DOUBLE:
+ append_double(groups, row, value.get_double());
+ return;
+ case VariantPrimitiveId::DECIMAL4:
+ case VariantPrimitiveId::DECIMAL8:
+ case VariantPrimitiveId::DECIMAL16:
+ append_decimal(groups, row, value.get_decimal());
+ return;
+ case VariantPrimitiveId::DATE:
+ append_date(groups, row, value.get_date());
+ return;
+ case VariantPrimitiveId::TIMESTAMP_MICROS:
+ append_timestamp(groups, row, value.get_timestamp_micros(), true);
+ return;
+ case VariantPrimitiveId::TIMESTAMP_NTZ_MICROS:
+ append_timestamp(groups, row, value.get_timestamp_ntz_micros(), false);
+ return;
+ case VariantPrimitiveId::TIMESTAMP_NANOS:
+ append_timestamp(groups, row,
nanos_to_micros(value.get_timestamp_nanos()), true);
+ return;
+ case VariantPrimitiveId::TIMESTAMP_NTZ_NANOS:
+ append_timestamp(groups, row,
nanos_to_micros(value.get_timestamp_ntz_nanos()), false);
+ return;
+ case VariantPrimitiveId::STRING:
+ append_string(groups, row, value.get_string());
+ return;
+ }
+ throw Exception(ErrorCode::CORRUPTION, "Unknown Variant primitive id");
+}
+
+Status execute_non_strict_scalar_cast(FunctionContext* context, const
ColumnPtr& source,
+ const DataTypePtr& source_type,
+ const DataTypePtr& target_type, const
char* source_name,
+ size_t rows, ColumnPtr* output) {
+ if (context == nullptr) {
+ return Status::InvalidArgument("Variant V2 scalar CAST requires a
FunctionContext");
+ }
+ auto cast_context = context->clone();
+ cast_context->set_enable_strict_mode(false);
+ const DataTypePtr nullable_target = make_nullable(target_type);
+ Block temporary {{source, source_type, source_name},
+ {nullable_target->create_column(), nullable_target,
"variant-v2-result"}};
+ WrapperType wrapper =
+ prepare_unpack_dictionaries(cast_context.get(), source_type,
nullable_target);
+ Status status = wrapper(cast_context.get(), temporary, {0}, 1, rows,
nullptr);
+ if (!status.ok()) {
+ if (status.is<ErrorCode::INVALID_ARGUMENT>()) {
+ *output = make_all_null_column(target_type, rows);
+ return Status::OK();
+ }
+ return status;
+ }
+ *output = make_nullable(temporary.get_by_position(1).column);
+ return Status::OK();
+}
+
+Status execute_concrete_cast(FunctionContext* context, const ScalarGroup&
group,
+ const DataTypePtr& target_type, ColumnPtr*
output) {
+ return execute_non_strict_scalar_cast(context, group.values->get_ptr(),
group.type, target_type,
+ "variant-v2-group",
group.source_rows.size(), output);
+}
+
+Status assemble_groups(FunctionContext* context, const ScalarGroups& groups,
+ const DataTypePtr& target_type, size_t rows, ColumnPtr*
output) {
+ const ScalarGroup* only_group = nullptr;
+ size_t only_group_index = 0;
+ for (size_t index = 0; index < groups.size(); ++index) {
+ if (groups[index].source_rows.empty()) {
+ continue;
+ }
+ if (only_group != nullptr) {
+ only_group = nullptr;
+ break;
+ }
+ only_group = &groups[index];
+ only_group_index = index;
+ }
+ if (only_group != nullptr && only_group->source_rows.size() == rows) {
+ if (only_group_index == INVALID_GROUP) {
+ *output = make_all_null_column(target_type, rows);
+ return Status::OK();
+ }
+ return execute_concrete_cast(context, *only_group, target_type,
output);
+ }
+
+ MutableColumnPtr nested = target_type->create_column();
+ auto nulls = ColumnUInt8::create();
+ nested->reserve(rows);
+ nulls->reserve(rows);
+ IColumn::Permutation permutation(rows);
+ size_t concatenated_row = 0;
+
+ for (size_t index = 0; index < groups.size(); ++index) {
+ const ScalarGroup& group = groups[index];
+ if (group.source_rows.empty()) {
+ continue;
+ }
+ ColumnPtr cast_result;
+ if (index == INVALID_GROUP) {
+ cast_result = make_all_null_column(target_type,
group.source_rows.size());
+ } else {
+ RETURN_IF_ERROR(execute_concrete_cast(context, group, target_type,
&cast_result));
+ }
+ const auto& nullable = assert_cast<const
ColumnNullable&>(*cast_result);
+ nested->insert_range_from(nullable.get_nested_column(), 0,
group.source_rows.size());
+ nulls->insert_range_from(nullable.get_null_map_column(), 0,
group.source_rows.size());
+ for (size_t source_row : group.source_rows) {
+ permutation[source_row] = concatenated_row++;
+ }
+ }
+ if (concatenated_row != rows) {
+ return Status::InternalError("Variant V2 scalar grouping produced {}
rows, expected {}",
+ concatenated_row, rows);
+ }
+ ColumnPtr concatenated = ColumnNullable::create(std::move(nested),
std::move(nulls));
+ *output = concatenated->permute(permutation, rows);
+ return Status::OK();
+}
+
+Status execute_typed_cast(FunctionContext* context, const ColumnPtr& source,
+ const DataTypePtr& source_type, const DataTypePtr&
target_type,
+ size_t rows, ColumnPtr* output) {
+ const DataTypePtr nullable_source = make_nullable(source_type);
+ return execute_non_strict_scalar_cast(context, source, nullable_source,
target_type,
+ "variant-v2-typed", rows, output);
+}
+
+} // namespace
+
+bool is_supported_scalar_source(const DataTypePtr& type) {
+ // Every scalar admitted by the typed state has a Parquet Variant mapping
in
+ // with_variant_typed_scalar(), so CAST and the physical typed-state
whitelist must stay
+ // identical.
+ return
is_supported_variant_typed_identity(remove_nullable(type)->get_primitive_type());
+}
+
+bool is_supported_scalar_target(const DataTypePtr& type) {
+ switch (remove_nullable(type)->get_primitive_type()) {
+ case TYPE_BOOLEAN:
+ case TYPE_TINYINT:
+ case TYPE_SMALLINT:
+ case TYPE_INT:
+ case TYPE_BIGINT:
+ case TYPE_LARGEINT:
+ case TYPE_FLOAT:
+ case TYPE_DOUBLE:
+ case TYPE_DECIMALV2:
+ case TYPE_DECIMAL32:
+ case TYPE_DECIMAL64:
+ case TYPE_DECIMAL128I:
+ case TYPE_DECIMAL256:
+ case TYPE_DATE:
+ case TYPE_DATEV2:
+ case TYPE_DATETIME:
+ case TYPE_DATETIMEV2:
+ case TYPE_TIMESTAMPTZ:
+ return true;
+ default:
+ return false;
+ }
+}
+
+Status cast_scalar_to_variant(const ColumnPtr& source, const DataTypePtr&
source_type, size_t rows,
+ ForcedNulls forced_nulls, ColumnPtr* output) {
+ if (!source || source->size() != rows ||
+ (!forced_nulls.empty() && forced_nulls.size() != rows)) {
+ return Status::InvalidArgument("Invalid scalar input shape for Variant
V2 CAST");
+ }
+ auto nulls = ColumnUInt8::create(rows, 0);
+ if (!forced_nulls.empty()) {
+ std::ranges::copy(forced_nulls, nulls->get_data().begin());
+ }
+ ColumnPtr null_map = std::move(nulls);
+ ColumnPtr nullable = ColumnNullable::create(source, null_map);
+ *output = ColumnVariantV2::create_typed(std::move(nullable), source_type);
Review Comment:
[P1] Validate STRING input before publishing a typed Variant. Doris STRING
can contain arbitrary bytes, but this path only wraps the source column;
`CAST(CAST(unhex('FF') AS STRING) AS VARIANT)` can therefore cast straight back
to STRING successfully, while grouping, serialization, exchange, or Variant
output later constructs `VariantScalarRef::string` and rejects the same row as
invalid UTF-8. That makes cast success depend on which operator first
materializes the Variant. Please validate visible non-null STRING rows at this
boundary (or require an explicit binary representation) and cover direct
output, round-trip, hash/group, and exchange behavior for invalid UTF-8.
--
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]
