yiguolei commented on code in PR #65561:
URL: https://github.com/apache/doris/pull/65561#discussion_r3618782861
##########
be/src/core/value/variant/variant_metadata.h:
##########
@@ -17,31 +17,36 @@
#pragma once
-#include <gen_cpp/Types_types.h>
-
-#include <atomic>
+#include <cstddef>
#include <cstdint>
-#include <mutex>
-namespace doris::io {
+#include "core/string_ref.h"
+
+namespace doris {
-class RemoteScanCacheWriteLimiter {
-public:
- RemoteScanCacheWriteLimiter(TUniqueId query_id, int64_t threshold_bytes);
+struct VariantMetadataRef {
Review Comment:
我们的代码什么时候用class，什么时候用struct？
##########
be/src/core/data_type_serde/data_type_variant_v2_serde_binary.cpp:
##########
@@ -0,0 +1,486 @@
+// 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 <limits>
+#include <new>
+#include <string_view>
+
+#include "common/exception.h"
+#include "core/column/column_const.h"
+#include "core/data_type_serde/data_type_variant_v2_serde.h"
+#include "core/data_type_serde/data_type_variant_v2_serde_binary_internal.h"
+#include "core/value/variant/variant_field.h"
+
+namespace doris::variant_v2_serde_binary_internal {
Review Comment:
这个binary 是啥含义？
##########
be/src/core/value/variant/variant_tracked_storage.h:
##########
@@ -15,13 +15,21 @@
// specific language governing permissions and limitations
Review Comment:
这为啥有一个单独的文件？
##########
be/src/core/value/variant/variant_scalar_encoding.cpp:
##########
@@ -0,0 +1,404 @@
+// 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_scalar_encoding.h"
+
+#include <bit>
+#include <cstring>
+#include <limits>
+
+#include "common/exception.h"
+#include "core/value/variant/variant_encoding.h"
+#include "util/utf8_check.h"
+
+namespace doris {
+namespace {
+
+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();
+
+unsigned __int128 magnitude(__int128 value) {
+ const auto unsigned_value = static_cast<unsigned __int128>(value);
+ return value < 0 ? ~unsigned_value + 1 : unsigned_value;
+}
+
+constexpr uint8_t primitive_header(VariantPrimitiveId id) noexcept {
+ return static_cast<uint8_t>(static_cast<uint8_t>(id) <<
VARIANT_VALUE_HEADER_SHIFT);
+}
+
+constexpr VariantPrimitiveId boolean_primitive_id(bool value) noexcept {
+ return value ? VariantPrimitiveId::TRUE_VALUE :
VariantPrimitiveId::FALSE_VALUE;
+}
+
+struct IntegerEncoding {
+ VariantPrimitiveId id;
+ uint8_t width;
+};
+
+IntegerEncoding resolve_integer_encoding(int64_t value, uint8_t
requested_width) {
+ uint8_t width = requested_width;
+ if (width == 0) {
+ if (value >= std::numeric_limits<int8_t>::min() &&
+ value <= std::numeric_limits<int8_t>::max()) {
+ width = 1;
+ } else if (value >= std::numeric_limits<int16_t>::min() &&
+ value <= std::numeric_limits<int16_t>::max()) {
+ width = 2;
+ } else if (value >= std::numeric_limits<int32_t>::min() &&
+ value <= std::numeric_limits<int32_t>::max()) {
+ width = 4;
+ } else {
+ width = 8;
+ }
+ }
+
+ VariantPrimitiveId id = VariantPrimitiveId::INT64;
+ bool fits = true;
+ if (width == 1) {
+ id = VariantPrimitiveId::INT8;
+ fits = value >= std::numeric_limits<int8_t>::min() &&
+ value <= std::numeric_limits<int8_t>::max();
+ } else if (width == 2) {
+ id = VariantPrimitiveId::INT16;
+ fits = value >= std::numeric_limits<int16_t>::min() &&
+ value <= std::numeric_limits<int16_t>::max();
+ } else if (width == 4) {
+ id = VariantPrimitiveId::INT32;
+ fits = value >= std::numeric_limits<int32_t>::min() &&
+ value <= std::numeric_limits<int32_t>::max();
+ } else if (width != 8) {
+ throw Exception(ErrorCode::INVALID_ARGUMENT,
+ "Variant integer width {} must be one of 1, 2, 4, or
8", width);
+ }
+ if (!fits) {
+ throw Exception(ErrorCode::INVALID_ARGUMENT,
+ "Variant integer value does not fit requested width
{}", width);
+ }
+ return {.id = id, .width = width};
+}
+
+void write_unsigned(char*& output, uint64_t value, uint8_t width) noexcept {
+ for (uint8_t byte = 0; byte < width; ++byte) {
+ *output++ = static_cast<char>(value >> (byte * 8));
+ }
+}
+
+void write_signed128(char*& output, __int128 value, uint8_t width) noexcept {
+ const auto unsigned_value = static_cast<unsigned __int128>(value);
+ for (uint8_t byte = 0; byte < width; ++byte) {
+ *output++ = static_cast<char>(unsigned_value >> (byte * 8));
+ }
+}
+
+void validate_string_ref(StringRef value, const char* description) {
+ if (value.data == nullptr && value.size != 0) {
+ throw Exception(ErrorCode::INVALID_ARGUMENT, "Variant {} has a null
data pointer",
+ description);
+ }
+}
+
+void validate_utf8_string_ref(StringRef value, const char* description) {
+ validate_string_ref(value, description);
+ if (value.size != 0 && !validate_utf8(value.data, value.size)) {
+ throw Exception(ErrorCode::INVALID_ARGUMENT, "Variant {} is not valid
UTF-8", description);
+ }
+}
+
+} // namespace
+
+VariantScalarEncodingPlan VariantScalarEncodingPlan::null_value() noexcept {
+ VariantScalarEncodingPlan plan;
+ plan._header = primitive_header(VariantPrimitiveId::NULL_VALUE);
+ plan._encoded_size = 1;
+ return plan;
+}
+
+VariantScalarEncodingPlan VariantScalarEncodingPlan::boolean(bool value)
noexcept {
+ VariantScalarEncodingPlan plan;
+ plan._header = primitive_header(boolean_primitive_id(value));
+ plan._encoded_size = 1;
+ return plan;
+}
+
+VariantScalarEncodingPlan VariantScalarEncodingPlan::integer(int64_t value,
uint8_t width) {
+ const IntegerEncoding encoding = resolve_integer_encoding(value, width);
+
+ VariantScalarEncodingPlan plan;
+ plan._header = primitive_header(encoding.id);
+ plan._unsigned_value = std::bit_cast<uint64_t>(value);
+ plan._payload_width = encoding.width;
+ plan._encoded_size = static_cast<size_t>(encoding.width) + 1;
+ plan._kind = PayloadKind::UNSIGNED;
+ return plan;
+}
+
+VariantScalarEncodingPlan VariantScalarEncodingPlan::float32(float value)
noexcept {
+ VariantScalarEncodingPlan plan;
+ plan._header = static_cast<uint8_t>(VariantPrimitiveId::FLOAT) <<
VARIANT_VALUE_HEADER_SHIFT;
+ plan._unsigned_value = std::bit_cast<uint32_t>(value);
+ plan._payload_width = sizeof(float);
+ plan._encoded_size = sizeof(float) + 1;
+ plan._kind = PayloadKind::UNSIGNED;
+ return plan;
+}
+
+VariantScalarEncodingPlan VariantScalarEncodingPlan::float64(double value)
noexcept {
+ VariantScalarEncodingPlan plan;
+ plan._header = static_cast<uint8_t>(VariantPrimitiveId::DOUBLE) <<
VARIANT_VALUE_HEADER_SHIFT;
+ plan._unsigned_value = std::bit_cast<uint64_t>(value);
+ plan._payload_width = sizeof(double);
+ plan._encoded_size = sizeof(double) + 1;
+ plan._kind = PayloadKind::UNSIGNED;
+ return plan;
+}
+
+VariantScalarEncodingPlan VariantScalarEncodingPlan::decimal(__int128
unscaled, uint8_t scale,
+ uint8_t width) {
+ if (scale > 38) {
+ throw Exception(ErrorCode::INVALID_ARGUMENT, "Variant decimal scale {}
is outside [0, 38]",
+ scale);
+ }
+ const unsigned __int128 absolute = magnitude(unscaled);
+ if (absolute > MAX_DECIMAL38) {
+ throw Exception(ErrorCode::INVALID_ARGUMENT,
+ "Variant decimal unscaled value exceeds precision 38");
+ }
+ if (width == 0) {
+ if (absolute <= DECIMAL4_MAX) {
+ width = 4;
+ } else if (absolute <= DECIMAL8_MAX) {
+ width = 8;
+ } else {
+ width = 16;
+ }
+ }
+
+ VariantPrimitiveId id = VariantPrimitiveId::DECIMAL16;
+ unsigned __int128 width_max = MAX_DECIMAL38;
+ if (width == 4) {
+ id = VariantPrimitiveId::DECIMAL4;
+ width_max = DECIMAL4_MAX;
+ } else if (width == 8) {
+ id = VariantPrimitiveId::DECIMAL8;
+ width_max = DECIMAL8_MAX;
+ } else if (width != 16) {
+ throw Exception(ErrorCode::INVALID_ARGUMENT,
+ "Variant decimal width {} must be one of 4, 8, or 16",
width);
+ }
+ if (absolute > width_max) {
+ throw Exception(ErrorCode::INVALID_ARGUMENT,
+ "Variant decimal unscaled value exceeds precision for
width {}", width);
+ }
+
+ VariantScalarEncodingPlan plan;
+ plan._header = static_cast<uint8_t>(static_cast<uint8_t>(id) <<
VARIANT_VALUE_HEADER_SHIFT);
+ plan._signed_value = unscaled;
+ plan._payload_width = width;
+ plan._scale = scale;
+ plan._encoded_size = static_cast<size_t>(width) + 2;
+ plan._kind = PayloadKind::SIGNED;
+ plan._has_scale = true;
+ return plan;
+}
+
+VariantScalarEncodingPlan VariantScalarEncodingPlan::date(int32_t
days_since_epoch) noexcept {
+ VariantScalarEncodingPlan plan;
+ plan._header = static_cast<uint8_t>(VariantPrimitiveId::DATE) <<
VARIANT_VALUE_HEADER_SHIFT;
+ plan._unsigned_value = std::bit_cast<uint32_t>(days_since_epoch);
+ plan._payload_width = sizeof(int32_t);
+ plan._encoded_size = sizeof(int32_t) + 1;
+ plan._kind = PayloadKind::UNSIGNED;
+ return plan;
+}
+
+VariantScalarEncodingPlan VariantScalarEncodingPlan::timestamp_micros(int64_t
value,
+ bool
utc_adjusted) noexcept {
+ VariantScalarEncodingPlan plan;
+ plan._header = static_cast<uint8_t>(
+ static_cast<uint8_t>(utc_adjusted ?
VariantPrimitiveId::TIMESTAMP_MICROS
+ :
VariantPrimitiveId::TIMESTAMP_NTZ_MICROS)
+ << VARIANT_VALUE_HEADER_SHIFT);
+ plan._unsigned_value = std::bit_cast<uint64_t>(value);
+ plan._payload_width = sizeof(int64_t);
+ plan._encoded_size = sizeof(int64_t) + 1;
+ plan._kind = PayloadKind::UNSIGNED;
+ return plan;
+}
+
+VariantScalarEncodingPlan VariantScalarEncodingPlan::timestamp_nanos(int64_t
value,
+ bool
utc_adjusted) noexcept {
+ VariantScalarEncodingPlan plan;
+ plan._header = static_cast<uint8_t>(
+ static_cast<uint8_t>(utc_adjusted ?
VariantPrimitiveId::TIMESTAMP_NANOS
+ :
VariantPrimitiveId::TIMESTAMP_NTZ_NANOS)
+ << VARIANT_VALUE_HEADER_SHIFT);
+ plan._unsigned_value = std::bit_cast<uint64_t>(value);
+ plan._payload_width = sizeof(int64_t);
+ plan._encoded_size = sizeof(int64_t) + 1;
+ plan._kind = PayloadKind::UNSIGNED;
+ return plan;
+}
+
+VariantScalarEncodingPlan VariantScalarEncodingPlan::time_ntz_micros(int64_t
value) noexcept {
+ VariantScalarEncodingPlan plan;
+ plan._header = static_cast<uint8_t>(VariantPrimitiveId::TIME_NTZ_MICROS)
+ << VARIANT_VALUE_HEADER_SHIFT;
+ plan._unsigned_value = std::bit_cast<uint64_t>(value);
+ plan._payload_width = sizeof(int64_t);
+ plan._encoded_size = sizeof(int64_t) + 1;
+ plan._kind = PayloadKind::UNSIGNED;
+ return plan;
+}
+
+VariantScalarEncodingPlan VariantScalarEncodingPlan::binary(StringRef value) {
+ validate_string_ref(value, "binary");
+ if (value.size > std::numeric_limits<uint32_t>::max()) {
+ throw Exception(ErrorCode::INVALID_ARGUMENT,
+ "Variant binary exceeds the uint32 byte limit");
+ }
+ VariantScalarEncodingPlan plan;
+ plan._header = static_cast<uint8_t>(VariantPrimitiveId::BINARY) <<
VARIANT_VALUE_HEADER_SHIFT;
+ plan._borrowed = value;
+ plan._encoded_size = value.size + sizeof(uint32_t) + 1;
+ plan._kind = PayloadKind::BORROWED_BYTES;
+ plan._has_length = true;
+ return plan;
+}
+
+VariantScalarEncodingPlan VariantScalarEncodingPlan::string(StringRef value) {
+ validate_utf8_string_ref(value, "string");
+ if (value.size > std::numeric_limits<uint32_t>::max()) {
+ throw Exception(ErrorCode::INVALID_ARGUMENT,
+ "Variant string exceeds the uint32 byte limit");
+ }
+ VariantScalarEncodingPlan plan;
+ plan._borrowed = value;
+ plan._kind = PayloadKind::BORROWED_BYTES;
+ if (value.size <= VARIANT_MAX_SHORT_STRING_SIZE) {
+ plan._header = static_cast<uint8_t>((value.size <<
VARIANT_VALUE_HEADER_SHIFT) |
+
static_cast<uint8_t>(VariantBasicType::SHORT_STRING));
+ plan._encoded_size = value.size + 1;
+ } else {
+ plan._header = static_cast<uint8_t>(VariantPrimitiveId::STRING)
+ << VARIANT_VALUE_HEADER_SHIFT;
+ plan._encoded_size = value.size + sizeof(uint32_t) + 1;
+ plan._has_length = true;
+ }
+ return plan;
+}
+
+VariantScalarEncodingPlan VariantScalarEncodingPlan::uuid(
Review Comment:
？？
##########
be/src/core/value/variant/variant_scalar_encoding.cpp:
##########
@@ -0,0 +1,404 @@
+// 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_scalar_encoding.h"
+
+#include <bit>
+#include <cstring>
+#include <limits>
+
+#include "common/exception.h"
+#include "core/value/variant/variant_encoding.h"
+#include "util/utf8_check.h"
+
+namespace doris {
+namespace {
+
+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();
+
+unsigned __int128 magnitude(__int128 value) {
+ const auto unsigned_value = static_cast<unsigned __int128>(value);
+ return value < 0 ? ~unsigned_value + 1 : unsigned_value;
+}
+
+constexpr uint8_t primitive_header(VariantPrimitiveId id) noexcept {
+ return static_cast<uint8_t>(static_cast<uint8_t>(id) <<
VARIANT_VALUE_HEADER_SHIFT);
+}
+
+constexpr VariantPrimitiveId boolean_primitive_id(bool value) noexcept {
+ return value ? VariantPrimitiveId::TRUE_VALUE :
VariantPrimitiveId::FALSE_VALUE;
+}
+
+struct IntegerEncoding {
+ VariantPrimitiveId id;
+ uint8_t width;
+};
+
+IntegerEncoding resolve_integer_encoding(int64_t value, uint8_t
requested_width) {
+ uint8_t width = requested_width;
+ if (width == 0) {
+ if (value >= std::numeric_limits<int8_t>::min() &&
+ value <= std::numeric_limits<int8_t>::max()) {
+ width = 1;
+ } else if (value >= std::numeric_limits<int16_t>::min() &&
+ value <= std::numeric_limits<int16_t>::max()) {
+ width = 2;
+ } else if (value >= std::numeric_limits<int32_t>::min() &&
+ value <= std::numeric_limits<int32_t>::max()) {
+ width = 4;
+ } else {
+ width = 8;
+ }
+ }
+
+ VariantPrimitiveId id = VariantPrimitiveId::INT64;
+ bool fits = true;
+ if (width == 1) {
+ id = VariantPrimitiveId::INT8;
+ fits = value >= std::numeric_limits<int8_t>::min() &&
+ value <= std::numeric_limits<int8_t>::max();
+ } else if (width == 2) {
+ id = VariantPrimitiveId::INT16;
+ fits = value >= std::numeric_limits<int16_t>::min() &&
+ value <= std::numeric_limits<int16_t>::max();
+ } else if (width == 4) {
+ id = VariantPrimitiveId::INT32;
+ fits = value >= std::numeric_limits<int32_t>::min() &&
+ value <= std::numeric_limits<int32_t>::max();
+ } else if (width != 8) {
+ throw Exception(ErrorCode::INVALID_ARGUMENT,
+ "Variant integer width {} must be one of 1, 2, 4, or
8", width);
+ }
+ if (!fits) {
+ throw Exception(ErrorCode::INVALID_ARGUMENT,
+ "Variant integer value does not fit requested width
{}", width);
+ }
+ return {.id = id, .width = width};
+}
+
+void write_unsigned(char*& output, uint64_t value, uint8_t width) noexcept {
+ for (uint8_t byte = 0; byte < width; ++byte) {
+ *output++ = static_cast<char>(value >> (byte * 8));
+ }
+}
+
+void write_signed128(char*& output, __int128 value, uint8_t width) noexcept {
+ const auto unsigned_value = static_cast<unsigned __int128>(value);
+ for (uint8_t byte = 0; byte < width; ++byte) {
+ *output++ = static_cast<char>(unsigned_value >> (byte * 8));
+ }
+}
+
+void validate_string_ref(StringRef value, const char* description) {
+ if (value.data == nullptr && value.size != 0) {
+ throw Exception(ErrorCode::INVALID_ARGUMENT, "Variant {} has a null
data pointer",
+ description);
+ }
+}
+
+void validate_utf8_string_ref(StringRef value, const char* description) {
+ validate_string_ref(value, description);
+ if (value.size != 0 && !validate_utf8(value.data, value.size)) {
+ throw Exception(ErrorCode::INVALID_ARGUMENT, "Variant {} is not valid
UTF-8", description);
+ }
+}
+
+} // namespace
+
+VariantScalarEncodingPlan VariantScalarEncodingPlan::null_value() noexcept {
+ VariantScalarEncodingPlan plan;
+ plan._header = primitive_header(VariantPrimitiveId::NULL_VALUE);
+ plan._encoded_size = 1;
+ return plan;
+}
+
+VariantScalarEncodingPlan VariantScalarEncodingPlan::boolean(bool value)
noexcept {
+ VariantScalarEncodingPlan plan;
+ plan._header = primitive_header(boolean_primitive_id(value));
+ plan._encoded_size = 1;
+ return plan;
+}
+
+VariantScalarEncodingPlan VariantScalarEncodingPlan::integer(int64_t value,
uint8_t width) {
+ const IntegerEncoding encoding = resolve_integer_encoding(value, width);
+
+ VariantScalarEncodingPlan plan;
+ plan._header = primitive_header(encoding.id);
+ plan._unsigned_value = std::bit_cast<uint64_t>(value);
+ plan._payload_width = encoding.width;
+ plan._encoded_size = static_cast<size_t>(encoding.width) + 1;
+ plan._kind = PayloadKind::UNSIGNED;
+ return plan;
+}
+
+VariantScalarEncodingPlan VariantScalarEncodingPlan::float32(float value)
noexcept {
+ VariantScalarEncodingPlan plan;
+ plan._header = static_cast<uint8_t>(VariantPrimitiveId::FLOAT) <<
VARIANT_VALUE_HEADER_SHIFT;
+ plan._unsigned_value = std::bit_cast<uint32_t>(value);
+ plan._payload_width = sizeof(float);
+ plan._encoded_size = sizeof(float) + 1;
+ plan._kind = PayloadKind::UNSIGNED;
+ return plan;
+}
+
+VariantScalarEncodingPlan VariantScalarEncodingPlan::float64(double value)
noexcept {
+ VariantScalarEncodingPlan plan;
+ plan._header = static_cast<uint8_t>(VariantPrimitiveId::DOUBLE) <<
VARIANT_VALUE_HEADER_SHIFT;
+ plan._unsigned_value = std::bit_cast<uint64_t>(value);
+ plan._payload_width = sizeof(double);
+ plan._encoded_size = sizeof(double) + 1;
+ plan._kind = PayloadKind::UNSIGNED;
+ return plan;
+}
+
+VariantScalarEncodingPlan VariantScalarEncodingPlan::decimal(__int128
unscaled, uint8_t scale,
+ uint8_t width) {
+ if (scale > 38) {
+ throw Exception(ErrorCode::INVALID_ARGUMENT, "Variant decimal scale {}
is outside [0, 38]",
+ scale);
+ }
+ const unsigned __int128 absolute = magnitude(unscaled);
+ if (absolute > MAX_DECIMAL38) {
+ throw Exception(ErrorCode::INVALID_ARGUMENT,
+ "Variant decimal unscaled value exceeds precision 38");
+ }
+ if (width == 0) {
+ if (absolute <= DECIMAL4_MAX) {
+ width = 4;
+ } else if (absolute <= DECIMAL8_MAX) {
+ width = 8;
+ } else {
+ width = 16;
+ }
+ }
+
+ VariantPrimitiveId id = VariantPrimitiveId::DECIMAL16;
+ unsigned __int128 width_max = MAX_DECIMAL38;
+ if (width == 4) {
+ id = VariantPrimitiveId::DECIMAL4;
+ width_max = DECIMAL4_MAX;
+ } else if (width == 8) {
+ id = VariantPrimitiveId::DECIMAL8;
+ width_max = DECIMAL8_MAX;
+ } else if (width != 16) {
+ throw Exception(ErrorCode::INVALID_ARGUMENT,
+ "Variant decimal width {} must be one of 4, 8, or 16",
width);
+ }
+ if (absolute > width_max) {
+ throw Exception(ErrorCode::INVALID_ARGUMENT,
+ "Variant decimal unscaled value exceeds precision for
width {}", width);
+ }
+
+ VariantScalarEncodingPlan plan;
+ plan._header = static_cast<uint8_t>(static_cast<uint8_t>(id) <<
VARIANT_VALUE_HEADER_SHIFT);
+ plan._signed_value = unscaled;
+ plan._payload_width = width;
+ plan._scale = scale;
+ plan._encoded_size = static_cast<size_t>(width) + 2;
+ plan._kind = PayloadKind::SIGNED;
+ plan._has_scale = true;
+ return plan;
+}
+
+VariantScalarEncodingPlan VariantScalarEncodingPlan::date(int32_t
days_since_epoch) noexcept {
+ VariantScalarEncodingPlan plan;
+ plan._header = static_cast<uint8_t>(VariantPrimitiveId::DATE) <<
VARIANT_VALUE_HEADER_SHIFT;
+ plan._unsigned_value = std::bit_cast<uint32_t>(days_since_epoch);
+ plan._payload_width = sizeof(int32_t);
+ plan._encoded_size = sizeof(int32_t) + 1;
+ plan._kind = PayloadKind::UNSIGNED;
+ return plan;
+}
+
+VariantScalarEncodingPlan VariantScalarEncodingPlan::timestamp_micros(int64_t
value,
+ bool
utc_adjusted) noexcept {
+ VariantScalarEncodingPlan plan;
+ plan._header = static_cast<uint8_t>(
+ static_cast<uint8_t>(utc_adjusted ?
VariantPrimitiveId::TIMESTAMP_MICROS
+ :
VariantPrimitiveId::TIMESTAMP_NTZ_MICROS)
+ << VARIANT_VALUE_HEADER_SHIFT);
+ plan._unsigned_value = std::bit_cast<uint64_t>(value);
+ plan._payload_width = sizeof(int64_t);
+ plan._encoded_size = sizeof(int64_t) + 1;
+ plan._kind = PayloadKind::UNSIGNED;
+ return plan;
+}
+
+VariantScalarEncodingPlan VariantScalarEncodingPlan::timestamp_nanos(int64_t
value,
Review Comment:
我们有这个？？
##########
be/src/core/value/variant/variant_canonical.cpp:
##########
@@ -0,0 +1,1292 @@
+// 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_encoding.h"
+#include "core/value/variant/variant_field.h"
+#include "exec/common/sip_hash.h"
+#include "util/hash_util.hpp"
+#include "util/utf8_check.h"
+
+namespace doris {
+namespace {
+
+constexpr unsigned __int128 DECIMAL4_MAX = 999'999'999;
+constexpr unsigned __int128 DECIMAL8_MAX = 999'999'999'999'999'999;
+constexpr uint64_t CANONICAL_NAN_BITS = 0x7FF8000000000000ULL;
+constexpr size_t CANONICAL_SIZE_PREFIX = sizeof(uint32_t);
+
+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;
+}
+
+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();
+
+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;
+}
+
+unsigned __int128 magnitude(__int128 value) {
+ const auto unsigned_value = static_cast<unsigned __int128>(value);
+ return value < 0 ? ~unsigned_value + 1 : unsigned_value;
+}
+
+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 (magnitude(integer) <= MAX_DECIMAL38) {
+ 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 (magnitude(decimal.unscaled) > MAX_DECIMAL38) {
+ 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_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;
+}
+
+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 = magnitude(value);
+ if (absolute <= DECIMAL4_MAX) {
+ return 4;
+ }
+ if (absolute <= 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;
+ std::vector<PlanChild> children;
+ std::vector<StringRef> keys;
+};
+
+uint32_t build_plan_node(SerializePlan& plan, VariantRef value, uint32_t
depth) {
+ require_depth(depth);
+ require_exact_value(value);
+ if (plan.nodes.size() == std::numeric_limits<uint32_t>::max()) {
+ throw Exception(ErrorCode::INVALID_ARGUMENT,
+ "Variant canonical plan exceeds uint32 node limit");
+ }
+ const NormalizedValue normalized = normalize_value(value);
+ const auto node_index = static_cast<uint32_t>(plan.nodes.size());
+ PlanNode node;
+ node.normalized = normalized;
+ plan.nodes.push_back(node);
+ if (normalized.kind != CanonicalKind::OBJECT && normalized.kind !=
CanonicalKind::ARRAY) {
+ plan.nodes[node_index].encoded_size = scalar_encoded_size(normalized);
+ return node_index;
+ }
+
+ const uint32_t count = value.num_elements();
+ if (count > plan.children.max_size() - plan.children.size()) {
+ throw Exception(ErrorCode::INVALID_ARGUMENT,
+ "Variant canonical plan exceeds addressable child
count");
+ }
+ const size_t children_begin = plan.children.size();
+ plan.children.resize(children_begin + count);
+ plan.nodes[node_index].children_begin = children_begin;
+ plan.nodes[node_index].child_count = count;
+
+ StringRef previous;
+ for (uint32_t index = 0; index < count; ++index) {
+ VariantRef child;
+ StringRef key;
+ if (normalized.kind == CanonicalKind::OBJECT) {
+ const ObjectEntry entry = object_entry_at(value, index, previous,
index != 0);
+ child = entry.value;
+ key = entry.key;
+ if (plan.keys.size() == std::numeric_limits<uint32_t>::max()) {
+ throw Exception(ErrorCode::INVALID_ARGUMENT,
+ "Variant canonical dictionary exceeds uint32
key references");
+ }
+ plan.keys.push_back(key);
+ previous = key;
+ } else {
+ child = value.array_at(index);
+ }
+ const uint32_t child_node = build_plan_node(plan, child, depth + 1);
+ PlanChild& planned_child = plan.children[children_begin + index];
+ planned_child.node_index = child_node;
+ planned_child.key = key;
+ }
+ return node_index;
+}
+
+bool string_ref_less(StringRef left, StringRef right) {
+ return left.compare(right) < 0;
+}
+
+void finish_plan(SerializePlan& plan) {
+ std::ranges::sort(plan.keys, string_ref_less);
+ plan.keys.erase(std::ranges::unique(plan.keys).begin(), plan.keys.end());
+ if (plan.keys.size() > std::numeric_limits<uint32_t>::max()) {
+ throw Exception(ErrorCode::INVALID_ARGUMENT,
+ "Variant canonical dictionary exceeds uint32 key
limit");
+ }
+
+ for (PlanNode& node : plan.nodes) {
+ if (node.normalized.kind != CanonicalKind::OBJECT) {
+ continue;
+ }
+ uint32_t previous_id = 0;
+ for (uint32_t index = 0; index < node.child_count; ++index) {
+ PlanChild& child = plan.children[node.children_begin + index];
+ const auto position = std::ranges::lower_bound(plan.keys,
child.key, string_ref_less);
+ DCHECK(position != plan.keys.end() && *position == child.key);
+ child.field_id = static_cast<uint32_t>(position -
plan.keys.begin());
+ if (index != 0) {
+ DCHECK_LT(previous_id, child.field_id);
+ }
+ previous_id = child.field_id;
+ }
+ }
+
+ for (size_t node_position = plan.nodes.size(); node_position != 0;
--node_position) {
+ PlanNode& node = plan.nodes[node_position - 1];
+ if (node.normalized.kind != CanonicalKind::OBJECT &&
+ node.normalized.kind != CanonicalKind::ARRAY) {
+ continue;
+ }
+ node.count_width = node.child_count >
std::numeric_limits<uint8_t>::max() ? sizeof(uint32_t)
+
: sizeof(uint8_t);
+ uint32_t values_size = 0;
+ for (uint32_t index = 0; index < node.child_count; ++index) {
+ const size_t child_size =
+ plan.nodes[plan.children[node.children_begin +
index].node_index].encoded_size;
+ if (child_size > std::numeric_limits<uint32_t>::max() -
values_size) {
+ throw Exception(ErrorCode::INVALID_ARGUMENT,
+ "Variant canonical container values exceed
uint32 byte limit");
+ }
+ values_size += static_cast<uint32_t>(child_size);
+ }
+ node.values_size = values_size;
+ node.offset_width = minimum_unsigned_width(values_size);
+ if (node.normalized.kind == CanonicalKind::OBJECT) {
+ const uint32_t maximum_id =
+ node.child_count == 0
+ ? 0
+ : plan.children[node.children_begin +
node.child_count - 1].field_id;
+ node.id_width = minimum_unsigned_width(maximum_id);
+ }
+
+ uint64_t encoded_size = 1 + node.count_width + values_size;
+ encoded_size += (static_cast<uint64_t>(node.child_count) + 1) *
node.offset_width;
+ if (node.normalized.kind == CanonicalKind::OBJECT) {
+ encoded_size += static_cast<uint64_t>(node.child_count) *
node.id_width;
+ }
+ if (encoded_size > std::numeric_limits<size_t>::max()) {
+ throw Exception(ErrorCode::INVALID_ARGUMENT,
+ "Variant canonical value exceeds addressable
output size");
+ }
+ node.encoded_size = static_cast<size_t>(encoded_size);
+ }
+}
+
+void write_unsigned(char*& output, unsigned __int128 value, uint8_t width)
noexcept {
+ for (uint8_t index = 0; index < width; ++index) {
+ *output++ = static_cast<char>(value >> (index * 8));
+ }
+}
+
+void write_signed(char*& output, __int128 value, uint8_t width) noexcept {
+ write_unsigned(output, static_cast<unsigned __int128>(value), width);
+}
+
+void write_primitive_header(char*& output, VariantPrimitiveId id) noexcept {
+ *output++ = static_cast<char>(static_cast<uint8_t>(id) <<
VARIANT_VALUE_HEADER_SHIFT);
+}
+
+void write_exact_integer(__int128 value, char*& output) noexcept {
+ if (!fits_int64(value)) {
+ write_primitive_header(output, VariantPrimitiveId::DECIMAL16);
+ *output++ = 0;
+ write_signed(output, value, 16);
+ return;
+ }
+ const uint8_t width = minimum_integer_width(value);
+ VariantPrimitiveId id = VariantPrimitiveId::INT64;
+ if (width == 1) {
+ id = VariantPrimitiveId::INT8;
+ } else if (width == 2) {
+ id = VariantPrimitiveId::INT16;
+ } else if (width == 4) {
+ id = VariantPrimitiveId::INT32;
+ }
+ write_primitive_header(output, id);
+ write_signed(output, value, width);
+}
+
+void write_decimal(const NormalizedValue& value, char*& output) noexcept {
+ const uint8_t width = minimum_decimal_width(value.integer);
+ VariantPrimitiveId id = VariantPrimitiveId::DECIMAL16;
+ if (width == 4) {
+ id = VariantPrimitiveId::DECIMAL4;
+ } else if (width == 8) {
+ id = VariantPrimitiveId::DECIMAL8;
+ }
+ write_primitive_header(output, id);
+ *output++ = static_cast<char>(value.scale);
+ write_signed(output, value.integer, width);
+}
+
+void copy_bytes(StringRef bytes, char*& output) noexcept {
+ if (bytes.size != 0) {
+ std::memcpy(output, bytes.data, bytes.size);
+ output += bytes.size;
+ }
+}
+
+void write_string(StringRef string, char*& output) noexcept {
+ if (string.size <= VARIANT_MAX_SHORT_STRING_SIZE) {
+ *output++ = static_cast<char>((string.size <<
VARIANT_VALUE_HEADER_SHIFT) |
+
static_cast<uint8_t>(VariantBasicType::SHORT_STRING));
+ } else {
+ write_primitive_header(output, VariantPrimitiveId::STRING);
+ write_unsigned(output, string.size, sizeof(uint32_t));
+ }
+ copy_bytes(string, output);
+}
+
+void write_binary(StringRef binary, char*& output) noexcept {
+ write_primitive_header(output, VariantPrimitiveId::BINARY);
+ write_unsigned(output, binary.size, sizeof(uint32_t));
+ copy_bytes(binary, output);
+}
+
+void write_timestamp(const NormalizedValue& value, char*& output) noexcept {
+ const bool use_micros = value.integer % 1000 == 0;
+ const bool utc_adjusted = value.kind == CanonicalKind::TIMESTAMP_TZ;
+ VariantPrimitiveId id;
+ __int128 physical_value;
+ if (use_micros) {
+ id = utc_adjusted ? VariantPrimitiveId::TIMESTAMP_MICROS
+ : VariantPrimitiveId::TIMESTAMP_NTZ_MICROS;
+ physical_value = value.integer / 1000;
+ } else {
+ id = utc_adjusted ? VariantPrimitiveId::TIMESTAMP_NANOS
+ : VariantPrimitiveId::TIMESTAMP_NTZ_NANOS;
+ physical_value = value.integer;
+ }
+ DCHECK(fits_int64(physical_value));
+ write_primitive_header(output, id);
+ write_signed(output, physical_value, sizeof(int64_t));
+}
+
+void write_scalar(const NormalizedValue& value, char*& output) noexcept {
Review Comment:
我还是没搞懂这个canoical 这堆代码，再就是似乎它还有一堆write 方法，引入了一个新的编码机制？
##########
be/src/core/value/variant/variant_block_builder.h:
##########
@@ -0,0 +1,252 @@
+// 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 <cstddef>
+#include <cstdint>
+#include <memory>
+
+#include "core/string_ref.h"
+#include "core/value/variant/variant_encoded_block.h"
+#include "core/value/variant/variant_metadata.h"
+#include "core/value/variant/variant_tracked_storage.h"
+#include "core/value/variant/variant_value.h"
+
+namespace doris {
+
+class VariantScalarEncodingPlan;
+class VariantCollectionCore;
+
+// Internal dictionary owner for one VariantBlockBuilder encoding unit. Rows
collect temporary key
+// ids first; seal() fixes the sorted dictionary and id remap for the
completed block.
+class VariantMetadataBuilder {
+public:
+ VariantMetadataBuilder();
+ ~VariantMetadataBuilder();
+
+ VariantMetadataBuilder(const VariantMetadataBuilder&) = delete;
+ VariantMetadataBuilder& operator=(const VariantMetadataBuilder&) = delete;
+ VariantMetadataBuilder(VariantMetadataBuilder&&) = delete;
+ VariantMetadataBuilder& operator=(VariantMetadataBuilder&&) = delete;
+
+ uint32_t register_key(StringRef key);
+ void seal();
+
+ bool is_sealed() const noexcept;
+ size_t num_keys() const noexcept;
+ uint32_t final_id(uint32_t temporary_id) const;
+ StringRef encoded_metadata() const;
+ VariantMetadataRef metadata_ref() const;
+
+private:
+ friend class VariantBlockBuilder;
+ friend class VariantCollectionCore;
+
+ void _begin_row();
+ void _retain_key(uint32_t temporary_id) noexcept;
+ void _complete_row() noexcept;
+ void _abort_row(const uint32_t* temporary_ids, size_t count, bool
was_collecting) noexcept;
+ void _reserve_keys(size_t count);
+ VariantTrackedString _take_encoded_metadata() noexcept;
+ StringRef _temporary_key(uint32_t temporary_id) const noexcept;
+ size_t _key_capacity() const noexcept;
+ size_t _key_capacity_growths() const noexcept;
+
+ struct Impl;
+ std::unique_ptr<Impl> _impl;
+};
+
+// Collects a block through one stack-only active Row at a time. The
implementation owns one
+// metadata dictionary and one set of block-level
scalar/node/container/child/row-root arenas.
+// The shared scalar/node/container/child scratch for the complete encoding
unit stays within its
+// uint32 index/offset domain. Row and its scopes never own per-row heap
containers and must not
+// outlive this builder.
+class VariantBlockBuilder {
Review Comment:
variant blob builder; variant metadata buidler
##########
fe/fe-core/src/main/java/org/apache/doris/catalog/BuiltinScalarFunctions.java:
##########
@@ -397,6 +398,8 @@
import org.apache.doris.nereids.trees.expressions.functions.scalar.Nvl;
import org.apache.doris.nereids.trees.expressions.functions.scalar.Overlay;
import
org.apache.doris.nereids.trees.expressions.functions.scalar.ParseDataSize;
+import
org.apache.doris.nereids.trees.expressions.functions.scalar.ParseToVariant;
+import
org.apache.doris.nereids.trees.expressions.functions.scalar.ParseToVariantErrorToNull;
Review Comment:
这个要改成try parse啊
##########
be/src/core/value/variant/variant_metadata_builder.cpp:
##########
@@ -0,0 +1,294 @@
+// 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 <parallel_hashmap/phmap.h>
+
+#include <algorithm>
+#include <cstdint>
+#include <deque>
+#include <functional>
+#include <limits>
+#include <ranges>
+#include <utility>
+
+#include "common/exception.h"
+#include "core/value/variant/variant_block_builder.h"
+#include "core/value/variant/variant_encoded_block.h"
+#include "core/value/variant/variant_encoding.h"
+#include "core/value/variant/variant_tracked_storage.h"
+#include "util/utf8_check.h"
+
+namespace doris {
+namespace {
+
+constexpr uint32_t INVALID_INDEX = std::numeric_limits<uint32_t>::max();
+
+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 validate_utf8_string_ref(StringRef value, const char* description) {
+ if (value.data == nullptr && value.size != 0) {
+ throw Exception(ErrorCode::INVALID_ARGUMENT, "Variant {} has a null
data pointer",
+ description);
+ }
+ if (value.size != 0 && !validate_utf8(value.data, value.size)) {
+ throw Exception(ErrorCode::INVALID_ARGUMENT, "Variant {} is not valid
UTF-8", description);
+ }
+}
+
+template <typename String>
+void append_unsigned(String& output, uint64_t value, uint8_t width) {
+ for (uint8_t byte = 0; byte < width; ++byte) {
+ output.push_back(static_cast<char>(value >> (byte * 8)));
+ }
+}
+
+} // namespace
+
+struct VariantMetadataBuilder::Impl {
+ using KeyMapValue = std::pair<const StringRef, uint32_t>;
Review Comment:
这个的声明为啥在blob builder 里？
##########
be/src/exprs/aggregate/aggregate_function_uniq.cpp:
##########
@@ -36,6 +36,11 @@ AggregateFunctionPtr create_aggregate_function_uniq(const
std::string& name,
const DataTypePtr&
result_type,
const bool
result_is_nullable,
const
AggregateFunctionAttr& attr) {
+ if (argument_types.size() == 1 &&
+ remove_nullable(argument_types[0])->get_primitive_type() ==
TYPE_VARIANT) {
+ return
creator_without_type::create<AggregateFunctionUniqVariant>(argument_types,
+
result_is_nullable, attr);
Review Comment:
为什么variant 需要重写？ 不能直接加到后面的类型列表里
##########
be/src/core/value/variant/variant_block_builder.h:
##########
@@ -0,0 +1,252 @@
+// 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 <cstddef>
+#include <cstdint>
+#include <memory>
+
+#include "core/string_ref.h"
+#include "core/value/variant/variant_encoded_block.h"
+#include "core/value/variant/variant_metadata.h"
+#include "core/value/variant/variant_tracked_storage.h"
+#include "core/value/variant/variant_value.h"
+
+namespace doris {
+
+class VariantScalarEncodingPlan;
+class VariantCollectionCore;
+
+// Internal dictionary owner for one VariantBlockBuilder encoding unit. Rows
collect temporary key
+// ids first; seal() fixes the sorted dictionary and id remap for the
completed block.
+class VariantMetadataBuilder {
+public:
+ VariantMetadataBuilder();
+ ~VariantMetadataBuilder();
+
+ VariantMetadataBuilder(const VariantMetadataBuilder&) = delete;
+ VariantMetadataBuilder& operator=(const VariantMetadataBuilder&) = delete;
+ VariantMetadataBuilder(VariantMetadataBuilder&&) = delete;
+ VariantMetadataBuilder& operator=(VariantMetadataBuilder&&) = delete;
+
+ uint32_t register_key(StringRef key);
+ void seal();
+
+ bool is_sealed() const noexcept;
+ size_t num_keys() const noexcept;
+ uint32_t final_id(uint32_t temporary_id) const;
+ StringRef encoded_metadata() const;
+ VariantMetadataRef metadata_ref() const;
+
+private:
+ friend class VariantBlockBuilder;
+ friend class VariantCollectionCore;
+
+ void _begin_row();
+ void _retain_key(uint32_t temporary_id) noexcept;
+ void _complete_row() noexcept;
+ void _abort_row(const uint32_t* temporary_ids, size_t count, bool
was_collecting) noexcept;
+ void _reserve_keys(size_t count);
+ VariantTrackedString _take_encoded_metadata() noexcept;
+ StringRef _temporary_key(uint32_t temporary_id) const noexcept;
+ size_t _key_capacity() const noexcept;
+ size_t _key_capacity_growths() const noexcept;
+
+ struct Impl;
+ std::unique_ptr<Impl> _impl;
+};
+
+// Collects a block through one stack-only active Row at a time. The
implementation owns one
+// metadata dictionary and one set of block-level
scalar/node/container/child/row-root arenas.
+// The shared scalar/node/container/child scratch for the complete encoding
unit stays within its
+// uint32 index/offset domain. Row and its scopes never own per-row heap
containers and must not
+// outlive this builder.
+class VariantBlockBuilder {
+public:
+ struct ReserveHint {
+ size_t rows = 0;
+ size_t metadata_keys = 0;
+ size_t scalar_bytes = 0;
+ size_t nodes = 0;
+ size_t containers = 0;
+ size_t children = 0;
+ };
+
+#ifdef BE_TEST
+ struct TestCounters {
+ size_t metadata_capacity_growths = 0;
+ size_t scalar_capacity_growths = 0;
+ size_t node_capacity_growths = 0;
+ size_t container_capacity_growths = 0;
+ size_t child_capacity_growths = 0;
+ size_t scope_stack_capacity_growths = 0;
+ size_t object_id_scratch_capacity_growths = 0;
+ size_t key_reference_capacity_growths = 0;
+ size_t container_plan_capacity_growths = 0;
+ size_t planned_object_child_capacity_growths = 0;
+ size_t row_root_capacity_growths = 0;
+ size_t object_token_capacity_growths = 0;
+ size_t object_schema_hits = 0;
+ size_t object_schema_fallbacks = 0;
+ size_t object_plan_reuses = 0;
+ size_t object_plan_fallbacks = 0;
+ size_t metadata_unique_keys = 0;
+ size_t metadata_key_capacity = 0;
+ size_t scalar_byte_capacity = 0;
+ size_t node_capacity = 0;
+ size_t container_capacity = 0;
+ size_t child_capacity = 0;
+ size_t scope_stack_capacity = 0;
+ size_t object_id_scratch_capacity = 0;
+ size_t key_reference_capacity = 0;
+ size_t container_plan_capacity = 0;
+ size_t planned_object_child_capacity = 0;
+ size_t row_root_capacity = 0;
+ size_t previous_object_token_capacity = 0;
+ size_t pending_object_token_capacity = 0;
+
+ size_t total_capacity_growths() const noexcept {
+ return metadata_capacity_growths + scalar_capacity_growths +
node_capacity_growths +
+ container_capacity_growths + child_capacity_growths +
+ scope_stack_capacity_growths +
object_id_scratch_capacity_growths +
+ key_reference_capacity_growths +
container_plan_capacity_growths +
+ planned_object_child_capacity_growths +
row_root_capacity_growths +
+ object_token_capacity_growths;
+ }
+ };
+#endif
+
+ class Row {
+ public:
+ class ObjectScope {
+ public:
+ ObjectScope(const ObjectScope&) = delete;
+ ObjectScope& operator=(const ObjectScope&) = delete;
+ ObjectScope(ObjectScope&& other) noexcept;
+ ObjectScope& operator=(ObjectScope&& other) = delete;
+ ~ObjectScope() = default;
+
+ void add_key(StringRef key);
+ void finish();
+
+ private:
+ friend class Row;
+ ObjectScope(VariantBlockBuilder* builder, uint64_t generation,
uint32_t token) noexcept;
+
+ VariantBlockBuilder* _builder;
+ uint64_t _generation;
+ uint32_t _token;
+ };
+
+ class ArrayScope {
+ public:
+ ArrayScope(const ArrayScope&) = delete;
+ ArrayScope& operator=(const ArrayScope&) = delete;
+ ArrayScope(ArrayScope&& other) noexcept;
+ ArrayScope& operator=(ArrayScope&& other) = delete;
+ ~ArrayScope() = default;
+
+ void finish();
+
+ private:
+ friend class Row;
+ ArrayScope(VariantBlockBuilder* builder, uint64_t generation,
uint32_t token) noexcept;
+
+ VariantBlockBuilder* _builder;
+ uint64_t _generation;
+ uint32_t _token;
+ };
+
+ Row(const Row&) = delete;
+ Row& operator=(const Row&) = delete;
+ Row(Row&& other) noexcept;
+ Row& operator=(Row&& other) = delete;
+ ~Row();
+
+ void add_null();
+ void add_bool(bool value);
+ void add_int(int64_t value);
+ void add_float(float value);
+ void add_double(double value);
+ void add_decimal(__int128 unscaled, uint8_t scale);
+ void add_decimal(__int128 unscaled, uint8_t scale, uint8_t width);
+ void add_date(int32_t days_since_epoch);
+ void add_timestamp_micros(int64_t value, bool utc_adjusted);
+ void add_timestamp_nanos(int64_t value, bool utc_adjusted);
+ void add_time_ntz_micros(int64_t value);
+ void add_binary(StringRef value);
+ void add_string(StringRef value);
+ void add_uuid(const std::array<uint8_t, 16>& value);
+ void add_largeint(__int128 value);
+ void add_value(VariantRef value);
+
+ ObjectScope start_object();
+ ArrayScope start_array();
+
+ void finish();
+ void abort();
+ bool is_finished() const noexcept;
+
+ private:
+ friend class VariantBlockBuilder;
+ Row(VariantBlockBuilder* builder, uint64_t generation) noexcept;
Review Comment:
generation 是什么含义
--
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]
