nprasad2077/doris
1
0 Fork 0
mirror of https://github.com/apache/doris.git synced 2026-11-04 18:29:59 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
9cd2008070ef0aba4ec23062ba7d090832d495d7
doris/be/src/exec/common/variant_util.cpp
T
lihangyu 9cd2008070 [Refactor](Variant) add NestedGroup path metadata support (#62848) 
Add NestedGroup path metadata through Variant storage and query
planning.

- Write indexed NestedGroup paths into Variant subcolumn metadata so
top-array nested fields keep positional nulls.

- Teach ColumnVariant and DataTypeVariant to carry nested-group typed
paths through sparse/subcolumn IO, casting, element access, and
compaction.

- Align FE Variant type SQL, property analysis, and expression-to-thrift
translation with NestedGroup path planning.

- Regenerate nested_in_top_array expected output for the NestedGroup
positional layout.
2026-05-07 13:45:55 +08:00

2332 lines
101 KiB
C++
Raw Blame History

// 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 "exec/common/variant_util.h"
#include <assert.h>
#include <fmt/format.h>
#include <gen_cpp/FrontendService.h>
#include <gen_cpp/FrontendService_types.h>
#include <gen_cpp/HeartbeatService_types.h>
#include <gen_cpp/MasterService_types.h>
#include <gen_cpp/Status_types.h>
#include <gen_cpp/Types_types.h>
#include <glog/logging.h>
#include <rapidjson/document.h>
#include <rapidjson/stringbuffer.h>
#include <rapidjson/writer.h>
#include <simdjson/simdjson.h> // IWYU pragma: keep
#include <unicode/uchar.h>
#include <algorithm>
#include <cassert>
#include <cstddef>
#include <cstdint>
#include <cstring>
#include <list>
#include <memory>
#include <mutex>
#include <optional>
#include <ostream>
#include <ranges>
#include <set>
#include <stack>
#include <string>
#include <string_view>
#include <unordered_map>
#include <utility>
#include <vector>
#include "common/config.h"
#include "common/status.h"
#include "core/assert_cast.h"
#include "core/block/block.h"
#include "core/block/column_numbers.h"
#include "core/block/column_with_type_and_name.h"
#include "core/column/column.h"
#include "core/column/column_array.h"
#include "core/column/column_map.h"
#include "core/column/column_nullable.h"
#include "core/column/column_string.h"
#include "core/column/column_variant.h"
#include "core/data_type/data_type.h"
#include "core/data_type/data_type_array.h"
#include "core/data_type/data_type_factory.hpp"
#include "core/data_type/data_type_jsonb.h"
#include "core/data_type/data_type_nullable.h"
#include "core/data_type/data_type_string.h"
#include "core/data_type/data_type_variant.h"
#include "core/data_type/define_primitive_type.h"
#include "core/data_type/get_least_supertype.h"
#include "core/data_type/primitive_type.h"
#include "core/field.h"
#include "core/typeid_cast.h"
#include "core/types.h"
#include "exec/common/field_visitors.h"
#include "exec/common/sip_hash.h"
#include "exprs/function/function.h"
#include "exprs/function/simple_function_factory.h"
#include "exprs/function_context.h"
#include "exprs/json_functions.h"
#include "re2/re2.h"
#include "runtime/exec_env.h"
#include "runtime/runtime_state.h"
#include "storage/olap_common.h"
#include "storage/rowset/beta_rowset.h"
#include "storage/rowset/rowset.h"
#include "storage/rowset/rowset_fwd.h"
#include "storage/segment/segment_loader.h"
#include "storage/segment/variant/nested_group_path.h"
#include "storage/segment/variant/variant_column_reader.h"
#include "storage/segment/variant/variant_column_writer_impl.h"
#include "storage/tablet/tablet.h"
#include "storage/tablet/tablet_fwd.h"
#include "storage/tablet/tablet_schema.h"
#include "util/client_cache.h"
#include "util/defer_op.h"
#include "util/json/json_parser.h"
#include "util/json/path_in_data.h"
#include "util/json/simd_json_parser.h"
namespace doris::variant_util {
inline void append_escaped_regex_char(std::string* regex_output, char ch) {
switch (ch) {
case '.':
case '^':
case '$':
case '+':
case '*':
case '?':
case '(':
case ')':
case '|':
case '{':
case '}':
case '[':
case ']':
case '\\':
regex_output->push_back('\\');
regex_output->push_back(ch);
break;
default:
regex_output->push_back(ch);
break;
}
}
// Small LRU to cap compiled glob patterns
constexpr size_t kGlobRegexCacheCapacity = 256;
struct GlobRegexCacheEntry {
std::shared_ptr<RE2> re2;
std::list<std::string>::iterator lru_it;
};
static std::mutex g_glob_regex_cache_mutex;
static std::list<std::string> g_glob_regex_cache_lru;
static std::unordered_map<std::string, GlobRegexCacheEntry> g_glob_regex_cache;
std::shared_ptr<RE2> get_or_build_re2(const std::string& glob_pattern) {
{
std::lock_guard<std::mutex> lock(g_glob_regex_cache_mutex);
auto it = g_glob_regex_cache.find(glob_pattern);
if (it != g_glob_regex_cache.end()) {
g_glob_regex_cache_lru.splice(g_glob_regex_cache_lru.begin(), g_glob_regex_cache_lru,
it->second.lru_it);
return it->second.re2;
}
}
std::string regex_pattern;
Status st = glob_to_regex(glob_pattern, &regex_pattern);
if (!st.ok()) {
return nullptr;
}
auto compiled = std::make_shared<RE2>(regex_pattern);
if (!compiled->ok()) {
return nullptr;
}
{
std::lock_guard<std::mutex> lock(g_glob_regex_cache_mutex);
auto it = g_glob_regex_cache.find(glob_pattern);
if (it != g_glob_regex_cache.end()) {
g_glob_regex_cache_lru.splice(g_glob_regex_cache_lru.begin(), g_glob_regex_cache_lru,
it->second.lru_it);
return it->second.re2;
}
g_glob_regex_cache_lru.push_front(glob_pattern);
g_glob_regex_cache.emplace(glob_pattern,
GlobRegexCacheEntry {compiled, g_glob_regex_cache_lru.begin()});
if (g_glob_regex_cache.size() > kGlobRegexCacheCapacity) {
const std::string& evict_key = g_glob_regex_cache_lru.back();
g_glob_regex_cache.erase(evict_key);
g_glob_regex_cache_lru.pop_back();
}
}
return compiled;
}
// Convert a restricted glob pattern into a regex.
// Supported: '*', '?', '[...]', '\\' escape. Others are treated as literals.
Status glob_to_regex(const std::string& glob_pattern, std::string* regex_pattern) {
regex_pattern->clear();
regex_pattern->append("^");
bool is_escaped = false;
size_t pattern_length = glob_pattern.size();
for (size_t index = 0; index < pattern_length; ++index) {
char current_char = glob_pattern[index];
if (is_escaped) {
append_escaped_regex_char(regex_pattern, current_char);
is_escaped = false;
continue;
}
if (current_char == '\\') {
is_escaped = true;
continue;
}
if (current_char == '*') {
regex_pattern->append(".*");
continue;
}
if (current_char == '?') {
regex_pattern->append(".");
continue;
}
if (current_char == '[') {
size_t class_index = index + 1;
bool class_closed = false;
bool is_class_escaped = false;
std::string class_buffer;
if (class_index < pattern_length &&
(glob_pattern[class_index] == '!' || glob_pattern[class_index] == '^')) {
class_buffer.push_back('^');
++class_index;
}
for (; class_index < pattern_length; ++class_index) {
char class_char = glob_pattern[class_index];
if (is_class_escaped) {
class_buffer.push_back(class_char);
is_class_escaped = false;
continue;
}
if (class_char == '\\') {
is_class_escaped = true;
continue;
}
if (class_char == ']') {
class_closed = true;
break;
}
class_buffer.push_back(class_char);
}
if (!class_closed) {
return Status::InvalidArgument("Unclosed character class in glob pattern: {}",
glob_pattern);
}
regex_pattern->append("[");
regex_pattern->append(class_buffer);
regex_pattern->append("]");
index = class_index;
continue;
}
append_escaped_regex_char(regex_pattern, current_char);
}
if (is_escaped) {
append_escaped_regex_char(regex_pattern, '\\');
}
regex_pattern->append("$");
return Status::OK();
}
bool glob_match_re2(const std::string& glob_pattern, const std::string& candidate_path) {
auto compiled = get_or_build_re2(glob_pattern);
if (compiled == nullptr) {
return false;
}
return RE2::FullMatch(candidate_path, *compiled);
}
bool is_regular_ng_compaction_subpath(const PathInData& path) {
const std::string& relative_path = path.get_path();
return !relative_path.empty() && !path.has_nested_part() &&
!segment_v2::contains_nested_group_marker(relative_path) &&
!segment_v2::is_root_nested_group_path(relative_path) &&
relative_path != SPARSE_COLUMN_PATH &&
relative_path.find(DOC_VALUE_COLUMN_PATH) == std::string::npos;
}
bool should_keep_existing_ng_compaction_subcolumn(const TabletColumn& column) {
if (!column.is_extracted_column()) {
return false;
}
return is_regular_ng_compaction_subpath(column.path_info_ptr()->copy_pop_front());
}
TabletIndexes clone_tablet_indexes(const std::vector<const TabletIndex*>& indexes) {
TabletIndexes cloned_indexes;
cloned_indexes.reserve(indexes.size());
for (const auto* index : indexes) {
cloned_indexes.emplace_back(std::make_shared<TabletIndex>(*index));
}
return cloned_indexes;
}
void keep_existing_ng_compaction_subcolumn(const TabletSchemaSPtr& source_schema,
const TabletColumnPtr& extracted_column,
TabletSchemaSPtr& output_schema,
TabletSchema::PathsSetInfo& paths_set_info) {
DCHECK(extracted_column->is_extracted_column());
DCHECK(should_keep_existing_ng_compaction_subcolumn(*extracted_column));
auto relative_path = extracted_column->path_info_ptr()->copy_pop_front();
const std::string path = relative_path.get_path();
output_schema->append_column(*extracted_column);
auto indexes = clone_tablet_indexes(source_schema->inverted_indexs(*extracted_column));
if (extracted_column->path_info_ptr()->get_is_typed()) {
TabletSchema::SubColumnInfo sub_column_info {.column = *extracted_column,
.indexes = std::move(indexes)};
paths_set_info.typed_path_set.emplace(path, std::move(sub_column_info));
return;
}
paths_set_info.sub_path_set.emplace(path);
if (!indexes.empty()) {
paths_set_info.subcolumn_indexes.emplace(path, std::move(indexes));
}
}
using ExistingNgCompactionSubcolumns = std::unordered_map<int32_t, std::vector<TabletColumnPtr>>;
struct NestedGroupCompactionMaterializationPlan {
std::vector<TabletColumnPtr> preserved_regular_subcolumns;
std::unordered_set<PathInData, PathInData::Hash> materialized_regular_paths;
PathToDataTypes additional_regular_path_to_data_types;
};
bool should_preserve_existing_ng_compaction_subcolumns(
const TabletColumnPtr& column,
const std::unordered_map<int32_t, VariantExtendedInfo>& uid_to_variant_extended_info) {
const auto info_it = uid_to_variant_extended_info.find(column->unique_id());
return column->variant_enable_nested_group() ||
(info_it != uid_to_variant_extended_info.end() && info_it->second.has_nested_group);
}
std::unordered_set<int32_t> collect_ng_compaction_root_uids(
const TabletSchemaSPtr& target,
const std::unordered_map<int32_t, VariantExtendedInfo>& uid_to_variant_extended_info) {
std::unordered_set<int32_t> root_uids;
for (const TabletColumnPtr& column : target->columns()) {
if (column->is_variant_type() && should_preserve_existing_ng_compaction_subcolumns(
column, uid_to_variant_extended_info)) {
root_uids.insert(column->unique_id());
}
}
return root_uids;
}
ExistingNgCompactionSubcolumns collect_existing_ng_compaction_subcolumns(
const TabletSchemaSPtr& target, const std::unordered_set<int32_t>& ng_root_uids) {
ExistingNgCompactionSubcolumns uid_to_existing_subcolumns;
for (const TabletColumnPtr& column : target->columns()) {
if (!column->is_extracted_column() || !ng_root_uids.contains(column->parent_unique_id()) ||
!should_keep_existing_ng_compaction_subcolumn(*column)) {
continue;
}
uid_to_existing_subcolumns[column->parent_unique_id()].push_back(column);
}
return uid_to_existing_subcolumns;
}
NestedGroupCompactionMaterializationPlan build_nested_group_compaction_materialization_plan(
const std::vector<TabletColumnPtr>& existing_subcolumns,
const VariantExtendedInfo& extended_info) {
NestedGroupCompactionMaterializationPlan plan;
plan.preserved_regular_subcolumns = existing_subcolumns;
for (const auto& column : existing_subcolumns) {
plan.materialized_regular_paths.emplace(column->path_info_ptr()->copy_pop_front());
}
for (const auto& [path, data_types] : extended_info.path_to_data_types) {
if (!is_regular_ng_compaction_subpath(path) ||
plan.materialized_regular_paths.contains(path)) {
continue;
}
plan.materialized_regular_paths.emplace(path);
plan.additional_regular_path_to_data_types.emplace(path, data_types);
}
return plan;
}
void append_nested_group_compaction_columns(const TabletSchemaSPtr& target,
const TabletColumnPtr& column,
const NestedGroupCompactionMaterializationPlan& plan,
TabletSchemaSPtr& output_schema,
TabletSchema::PathsSetInfo& paths_set_info) {
for (const auto& existing_column : plan.preserved_regular_subcolumns) {
keep_existing_ng_compaction_subcolumn(target, existing_column, output_schema,
paths_set_info);
}
VariantCompactionUtil::get_compaction_subcolumns_from_data_types(
paths_set_info, column, target, plan.additional_regular_path_to_data_types,
output_schema);
}
size_t get_number_of_dimensions(const IDataType& type) {
if (const auto* type_array = typeid_cast<const DataTypeArray*>(&type)) {
return type_array->get_number_of_dimensions();
}
return 0;
}
size_t get_number_of_dimensions(const IColumn& column) {
if (const auto* column_array = check_and_get_column<ColumnArray>(column)) {
return column_array->get_number_of_dimensions();
}
return 0;
}
DataTypePtr get_base_type_of_array(const DataTypePtr& type) {
/// Get raw pointers to avoid extra copying of type pointers.
const DataTypeArray* last_array = nullptr;
const auto* current_type = type.get();
if (const auto* nullable = typeid_cast<const DataTypeNullable*>(current_type)) {
current_type = nullable->get_nested_type().get();
}
while (const auto* type_array = typeid_cast<const DataTypeArray*>(current_type)) {
current_type = type_array->get_nested_type().get();
last_array = type_array;
if (const auto* nullable = typeid_cast<const DataTypeNullable*>(current_type)) {
current_type = nullable->get_nested_type().get();
}
}
return last_array ? last_array->get_nested_type() : type;
}
Status cast_column(const ColumnWithTypeAndName& arg, const DataTypePtr& type, ColumnPtr* result) {
ColumnsWithTypeAndName arguments {arg, {nullptr, type, type->get_name()}};
// To prevent from null info lost, we should not call function since the function framework will wrap
// nullable to Variant instead of the root of Variant
// correct output: Nullable(Array(int)) -> Nullable(Variant(Nullable(Array(int))))
// incorrect output: Nullable(Array(int)) -> Nullable(Variant(Array(int)))
if (type->get_primitive_type() == TYPE_VARIANT) {
// If source column is variant, so the nullable info is different from dst column
if (arg.type->get_primitive_type() == TYPE_VARIANT) {
*result = type->is_nullable() ? make_nullable(arg.column) : remove_nullable(arg.column);
return Status::OK();
}
// set variant root column/type to from column/type
CHECK(arg.column->is_nullable());
auto to_type = remove_nullable(type);
const auto& data_type_object = assert_cast<const DataTypeVariant&>(*to_type);
auto variant = ColumnVariant::create(data_type_object.variant_max_subcolumns_count(),
data_type_object.enable_doc_mode());
variant->create_root(arg.type, arg.column->assume_mutable());
ColumnPtr nullable = ColumnNullable::create(
variant->get_ptr(),
check_and_get_column<ColumnNullable>(arg.column.get())->get_null_map_column_ptr());
*result = type->is_nullable() ? nullable : variant->get_ptr();
return Status::OK();
}
auto function = SimpleFunctionFactory::instance().get_function("CAST", arguments, type);
if (!function) {
return Status::InternalError("Not found cast function {} to {}", arg.type->get_name(),
type->get_name());
}
Block tmp_block {arguments};
uint32_t result_column = cast_set<uint32_t>(tmp_block.columns());
RuntimeState state;
auto ctx = FunctionContext::create_context(&state, {}, {});
if (arg.type->get_primitive_type() == INVALID_TYPE) {
// cast from nothing to any type should result in nulls
*result = type->create_column_const_with_default_value(arg.column->size())
->convert_to_full_column_if_const();
return Status::OK();
}
// We convert column string to jsonb type just add a string jsonb field to dst column instead of parse
// each line in original string column.
ctx->set_string_as_jsonb_string(true);
ctx->set_jsonb_string_as_string(true);
tmp_block.insert({nullptr, type, arg.name});
// TODO(lihangyu): we should handle this error in strict mode
if (!function->execute(ctx.get(), tmp_block, {0}, result_column, arg.column->size())) {
LOG_EVERY_N(WARNING, 100) << fmt::format("cast from {} to {}", arg.type->get_name(),
type->get_name());
*result = type->create_column_const_with_default_value(arg.column->size())
->convert_to_full_column_if_const();
return Status::OK();
}
*result = tmp_block.get_by_position(result_column).column->convert_to_full_column_if_const();
VLOG_DEBUG << fmt::format("{} before convert {}, after convert {}", arg.name,
arg.column->get_name(), (*result)->get_name());
return Status::OK();
}
void get_column_by_type(const DataTypePtr& data_type, const std::string& name, TabletColumn& column,
const ExtraInfo& ext_info) {
column.set_name(name);
column.set_type(data_type->get_storage_field_type());
if (ext_info.unique_id >= 0) {
column.set_unique_id(ext_info.unique_id);
}
if (ext_info.parent_unique_id >= 0) {
column.set_parent_unique_id(ext_info.parent_unique_id);
}
if (!ext_info.path_info.empty()) {
column.set_path_info(ext_info.path_info);
}
if (data_type->is_nullable()) {
const auto& real_type = static_cast<const DataTypeNullable&>(*data_type);
column.set_is_nullable(true);
get_column_by_type(real_type.get_nested_type(), name, column, {});
return;
}
if (data_type->get_primitive_type() == PrimitiveType::TYPE_ARRAY) {
TabletColumn child;
get_column_by_type(assert_cast<const DataTypeArray*>(data_type.get())->get_nested_type(),
"", child, {});
column.set_length(TabletColumn::get_field_length_by_type(TPrimitiveType::ARRAY, 0));
column.add_sub_column(child);
return;
}
if (data_type->get_primitive_type() == PrimitiveType::TYPE_VARIANT) {
const auto* dt_variant = assert_cast<const DataTypeVariant*>(data_type.get());
column.set_variant_max_subcolumns_count(dt_variant->variant_max_subcolumns_count());
column.set_variant_enable_doc_mode(dt_variant->enable_doc_mode());
return;
}
// size is not fixed when type is string or json
if (is_string_type(data_type->get_primitive_type()) ||
data_type->get_primitive_type() == TYPE_JSONB) {
column.set_length(INT_MAX);
return;
}
PrimitiveType type = data_type->get_primitive_type();
if (is_int_or_bool(type) || is_string_type(type) || is_float_or_double(type) || is_ip(type) ||
is_date_or_datetime(type) || type == PrimitiveType::TYPE_DATEV2) {
column.set_length(cast_set<int32_t>(data_type->get_size_of_value_in_memory()));
return;
}
if (is_decimal(type)) {
column.set_precision(data_type->get_precision());
column.set_frac(data_type->get_scale());
return;
}
// datetimev2 needs scale
if (type == PrimitiveType::TYPE_DATETIMEV2 || type == PrimitiveType::TYPE_TIMESTAMPTZ) {
column.set_precision(-1);
column.set_frac(data_type->get_scale());
return;
}
throw doris::Exception(doris::ErrorCode::INTERNAL_ERROR,
"unexcepted data column type: {}, column name is: {}",
data_type->get_name(), name);
}
TabletColumn get_column_by_type(const DataTypePtr& data_type, const std::string& name,
const ExtraInfo& ext_info) {
TabletColumn result;
get_column_by_type(data_type, name, result, ext_info);
return result;
}
// check if two paths which same prefix have different structure
static bool has_different_structure_in_same_path(const PathInData::Parts& lhs,
const PathInData::Parts& rhs) {
if (lhs.size() != rhs.size()) {
return false; // different size means different structure
}
// Since we group by path string, lhs and rhs must have the same size and keys
// We only need to check if they have different nested structure
for (size_t i = 0; i < lhs.size(); ++i) {
if (lhs[i] != rhs[i]) {
VLOG_DEBUG << fmt::format(
"Check different structure: {} vs {}, lhs[i].is_nested: {}, rhs[i].is_nested: "
"{}",
lhs[i].key, rhs[i].key, lhs[i].is_nested, rhs[i].is_nested);
return true;
}
}
return false;
}
Status check_variant_has_no_ambiguous_paths(const PathsInData& tuple_paths) {
// Group paths by their string representation to reduce comparisons
std::unordered_map<std::string, std::vector<size_t>> path_groups;
for (size_t i = 0; i < tuple_paths.size(); ++i) {
// same path should have same structure, so we group them by path
path_groups[tuple_paths[i].get_path()].push_back(i);
// print part of tuple_paths[i]
VLOG_DEBUG << "tuple_paths[i]: " << tuple_paths[i].get_path();
}
// Only compare paths within the same group
for (const auto& [path_str, indices] : path_groups) {
if (indices.size() <= 1) {
continue; // No conflicts possible
}
// Compare all pairs within this group
for (size_t i = 0; i < indices.size(); ++i) {
for (size_t j = 0; j < i; ++j) {
if (has_different_structure_in_same_path(tuple_paths[indices[i]].get_parts(),
tuple_paths[indices[j]].get_parts())) {
return Status::DataQualityError(
"Ambiguous paths: {} vs {} with different nested part {} vs {}",
tuple_paths[indices[i]].get_path(), tuple_paths[indices[j]].get_path(),
tuple_paths[indices[i]].has_nested_part(),
tuple_paths[indices[j]].has_nested_part());
}
}
}
}
return Status::OK();
}
Status update_least_schema_internal(const std::map<PathInData, DataTypes>& subcolumns_types,
TabletSchemaSPtr& common_schema, int32_t variant_col_unique_id,
const std::map<std::string, TabletColumnPtr>& typed_columns,
std::set<PathInData>* path_set) {
PathsInData tuple_paths;
DataTypes tuple_types;
CHECK(common_schema.use_count() == 1);
// Get the least common type for all paths.
for (const auto& [key, subtypes] : subcolumns_types) {
assert(!subtypes.empty());
if (key.get_path() == ColumnVariant::COLUMN_NAME_DUMMY) {
continue;
}
size_t first_dim = get_number_of_dimensions(*subtypes[0]);
tuple_paths.emplace_back(key);
for (size_t i = 1; i < subtypes.size(); ++i) {
if (first_dim != get_number_of_dimensions(*subtypes[i])) {
tuple_types.emplace_back(make_nullable(std::make_shared<DataTypeJsonb>()));
LOG(INFO) << fmt::format(
"Uncompatible types of subcolumn '{}': {} and {}, cast to JSONB",
key.get_path(), subtypes[0]->get_name(), subtypes[i]->get_name());
break;
}
}
if (tuple_paths.size() == tuple_types.size()) {
continue;
}
DataTypePtr common_type;
get_least_supertype_jsonb(subtypes, &common_type);
if (!common_type->is_nullable()) {
common_type = make_nullable(common_type);
}
tuple_types.emplace_back(common_type);
}
CHECK_EQ(tuple_paths.size(), tuple_types.size());
// Append all common type columns of this variant
for (int i = 0; i < tuple_paths.size(); ++i) {
TabletColumn common_column;
// typed path not contains root part
auto path_without_root = tuple_paths[i].copy_pop_front().get_path();
if (typed_columns.contains(path_without_root) && !tuple_paths[i].has_nested_part()) {
common_column = *typed_columns.at(path_without_root);
// parent unique id and path may not be init in write path
common_column.set_parent_unique_id(variant_col_unique_id);
common_column.set_path_info(tuple_paths[i]);
common_column.set_name(tuple_paths[i].get_path());
} else {
// const std::string& column_name = variant_col_name + "." + tuple_paths[i].get_path();
get_column_by_type(tuple_types[i], tuple_paths[i].get_path(), common_column,
ExtraInfo {.unique_id = -1,
.parent_unique_id = variant_col_unique_id,
.path_info = tuple_paths[i]});
}
common_schema->append_column(common_column);
if (path_set != nullptr) {
path_set->insert(tuple_paths[i]);
}
}
return Status::OK();
}
Status update_least_common_schema(const std::vector<TabletSchemaSPtr>& schemas,
TabletSchemaSPtr& common_schema, int32_t variant_col_unique_id,
std::set<PathInData>* path_set) {
std::map<std::string, TabletColumnPtr> typed_columns;
for (const TabletColumnPtr& col :
common_schema->column_by_uid(variant_col_unique_id).get_sub_columns()) {
typed_columns[col->name()] = col;
}
// Types of subcolumns by path from all tuples.
std::map<PathInData, DataTypes> subcolumns_types;
// Collect all paths first to enable batch checking
std::vector<PathInData> all_paths;
for (const TabletSchemaSPtr& schema : schemas) {
for (const TabletColumnPtr& col : schema->columns()) {
// Get subcolumns of this variant
if (col->has_path_info() && col->parent_unique_id() >= 0 &&
col->parent_unique_id() == variant_col_unique_id) {
subcolumns_types[*col->path_info_ptr()].emplace_back(
DataTypeFactory::instance().create_data_type(*col, col->is_nullable()));
all_paths.push_back(*col->path_info_ptr());
}
}
}
// Batch check for conflicts
RETURN_IF_ERROR(check_variant_has_no_ambiguous_paths(all_paths));
return update_least_schema_internal(subcolumns_types, common_schema, variant_col_unique_id,
typed_columns, path_set);
}
// Keep variant subcolumn BF support aligned with FE DDL checks.
bool is_bf_supported_by_fe_for_variant_subcolumn(FieldType type) {
switch (type) {
case FieldType::OLAP_FIELD_TYPE_SMALLINT:
case FieldType::OLAP_FIELD_TYPE_INT:
case FieldType::OLAP_FIELD_TYPE_BIGINT:
case FieldType::OLAP_FIELD_TYPE_LARGEINT:
case FieldType::OLAP_FIELD_TYPE_CHAR:
case FieldType::OLAP_FIELD_TYPE_VARCHAR:
case FieldType::OLAP_FIELD_TYPE_STRING:
case FieldType::OLAP_FIELD_TYPE_DATE:
case FieldType::OLAP_FIELD_TYPE_DATETIME:
case FieldType::OLAP_FIELD_TYPE_DATEV2:
case FieldType::OLAP_FIELD_TYPE_DATETIMEV2:
case FieldType::OLAP_FIELD_TYPE_TIMESTAMPTZ:
case FieldType::OLAP_FIELD_TYPE_DECIMAL:
case FieldType::OLAP_FIELD_TYPE_DECIMAL32:
case FieldType::OLAP_FIELD_TYPE_DECIMAL64:
case FieldType::OLAP_FIELD_TYPE_DECIMAL128I:
case FieldType::OLAP_FIELD_TYPE_DECIMAL256:
case FieldType::OLAP_FIELD_TYPE_IPV4:
case FieldType::OLAP_FIELD_TYPE_IPV6:
return true;
default:
return false;
}
}
void inherit_column_attributes(const TabletColumn& source, TabletColumn& target,
TabletSchemaSPtr* target_schema) {
if (!target.is_extracted_column()) {
return;
}
target.set_aggregation_method(source.aggregation());
// 1. bloom filter
if (is_bf_supported_by_fe_for_variant_subcolumn(target.type())) {
target.set_is_bf_column(source.is_bf_column());
}
if (!target_schema) {
return;
}
// 2. inverted index
TabletIndexes indexes_to_add;
auto source_indexes = (*target_schema)->inverted_indexs(source.unique_id());
// if target is variant type, we need to inherit all indexes
// because this schema is a read schema from fe
if (target.is_variant_type()) {
for (auto& index : source_indexes) {
auto index_info = std::make_shared<TabletIndex>(*index);
index_info->set_escaped_escaped_index_suffix_path(target.path_info_ptr()->get_path());
indexes_to_add.emplace_back(std::move(index_info));
}
} else {
inherit_index(source_indexes, indexes_to_add, target);
}
auto target_indexes = (*target_schema)
->inverted_indexs(target.parent_unique_id(),
target.path_info_ptr()->get_path());
if (target_indexes.empty()) {
for (auto& index_info : indexes_to_add) {
(*target_schema)->append_index(std::move(*index_info));
}
}
// 3. TODO: gnragm bf index
}
void inherit_column_attributes(TabletSchemaSPtr& schema) {
// Add index meta if extracted column is missing index meta
for (size_t i = 0; i < schema->num_columns(); ++i) {
TabletColumn& col = schema->mutable_column(i);
if (!col.is_extracted_column()) {
continue;
}
if (schema->field_index(col.parent_unique_id()) == -1) {
// parent column is missing, maybe dropped
continue;
}
inherit_column_attributes(schema->column_by_uid(col.parent_unique_id()), col, &schema);
}
}
Status get_least_common_schema(const std::vector<TabletSchemaSPtr>& schemas,
const TabletSchemaSPtr& base_schema, TabletSchemaSPtr& output_schema,
bool check_schema_size) {
std::vector<int32_t> variant_column_unique_id;
// Construct a schema excluding the extracted columns and gather unique identifiers for variants.
// Ensure that the output schema also excludes these extracted columns. This approach prevents
// duplicated paths following the update_least_common_schema process.
auto build_schema_without_extracted_columns = [&](const TabletSchemaSPtr& base_schema) {
output_schema = std::make_shared<TabletSchema>();
// not copy columns but only shadow copy other attributes
output_schema->shawdow_copy_without_columns(*base_schema);
// Get all columns without extracted columns and collect variant col unique id
for (const TabletColumnPtr& col : base_schema->columns()) {
if (col->is_variant_type()) {
variant_column_unique_id.push_back(col->unique_id());
}
if (!col->is_extracted_column()) {
output_schema->append_column(*col);
}
}
};
if (base_schema == nullptr) {
// Pick tablet schema with max schema version
auto max_version_schema =
*std::max_element(schemas.cbegin(), schemas.cend(),
[](const TabletSchemaSPtr a, const TabletSchemaSPtr b) {
return a->schema_version() < b->schema_version();
});
CHECK(max_version_schema);
build_schema_without_extracted_columns(max_version_schema);
} else {
// use input base_schema schema as base schema
build_schema_without_extracted_columns(base_schema);
}
for (int32_t unique_id : variant_column_unique_id) {
std::set<PathInData> path_set;
RETURN_IF_ERROR(update_least_common_schema(schemas, output_schema, unique_id, &path_set));
}
inherit_column_attributes(output_schema);
if (check_schema_size &&
output_schema->columns().size() > config::variant_max_merged_tablet_schema_size) {
return Status::DataQualityError("Reached max column size limit {}",
config::variant_max_merged_tablet_schema_size);
}
return Status::OK();
}
// sort by paths in lexicographical order
ColumnVariant::Subcolumns get_sorted_subcolumns(const ColumnVariant::Subcolumns& subcolumns) {
// sort by paths in lexicographical order
ColumnVariant::Subcolumns sorted = subcolumns;
std::sort(sorted.begin(), sorted.end(), [](const auto& lhsItem, const auto& rhsItem) {
return lhsItem->path < rhsItem->path;
});
return sorted;
}
bool has_schema_index_diff(const TabletSchema* new_schema, const TabletSchema* old_schema,
int32_t new_col_idx, int32_t old_col_idx) {
const auto& column_new = new_schema->column(new_col_idx);
const auto& column_old = old_schema->column(old_col_idx);
if (column_new.is_bf_column() != column_old.is_bf_column()) {
return true;
}
auto new_schema_inverted_indexs = new_schema->inverted_indexs(column_new);
auto old_schema_inverted_indexs = old_schema->inverted_indexs(column_old);
if (new_schema_inverted_indexs.size() != old_schema_inverted_indexs.size()) {
return true;
}
for (size_t i = 0; i < new_schema_inverted_indexs.size(); ++i) {
if (!new_schema_inverted_indexs[i]->is_same_except_id(old_schema_inverted_indexs[i])) {
return true;
}
}
return false;
}
TabletColumn create_sparse_column(const TabletColumn& variant) {
TabletColumn res;
res.set_name(variant.name_lower_case() + "." + SPARSE_COLUMN_PATH);
res.set_type(FieldType::OLAP_FIELD_TYPE_MAP);
res.set_aggregation_method(variant.aggregation());
res.set_path_info(PathInData {variant.name_lower_case() + "." + SPARSE_COLUMN_PATH});
res.set_parent_unique_id(variant.unique_id());
// set default value to "NULL" DefaultColumnIterator will call insert_many_defaults
res.set_default_value("NULL");
TabletColumn child_tcolumn;
child_tcolumn.set_type(FieldType::OLAP_FIELD_TYPE_STRING);
res.add_sub_column(child_tcolumn);
res.add_sub_column(child_tcolumn);
return res;
}
TabletColumn create_sparse_shard_column(const TabletColumn& variant, int bucket_index) {
TabletColumn res;
std::string name = variant.name_lower_case() + "." + SPARSE_COLUMN_PATH + ".b" +
std::to_string(bucket_index);
res.set_name(name);
res.set_type(FieldType::OLAP_FIELD_TYPE_MAP);
res.set_aggregation_method(variant.aggregation());
res.set_parent_unique_id(variant.unique_id());
res.set_default_value("NULL");
PathInData path(name);
res.set_path_info(path);
TabletColumn child_tcolumn;
child_tcolumn.set_type(FieldType::OLAP_FIELD_TYPE_STRING);
res.add_sub_column(child_tcolumn);
res.add_sub_column(child_tcolumn);
return res;
}
TabletColumn create_doc_value_column(const TabletColumn& variant, int bucket_index) {
TabletColumn res;
std::string name = variant.name_lower_case() + "." + DOC_VALUE_COLUMN_PATH + ".b" +
std::to_string(bucket_index);
res.set_name(name);
res.set_type(FieldType::OLAP_FIELD_TYPE_MAP);
res.set_aggregation_method(variant.aggregation());
res.set_parent_unique_id(variant.unique_id());
res.set_default_value("NULL");
res.set_path_info(PathInData {name});
TabletColumn child_tcolumn;
child_tcolumn.set_type(FieldType::OLAP_FIELD_TYPE_STRING);
res.add_sub_column(child_tcolumn);
res.add_sub_column(child_tcolumn);
return res;
}
uint32_t variant_binary_shard_of(const StringRef& path, uint32_t bucket_num) {
if (bucket_num <= 1) return 0;
SipHash hash;
hash.update(path.data, path.size);
uint64_t h = hash.get64();
return static_cast<uint32_t>(h % bucket_num);
}
Status VariantCompactionUtil::aggregate_path_to_stats(
const RowsetSharedPtr& rs,
std::unordered_map<int32_t, PathToNoneNullValues>* uid_to_path_stats) {
SegmentCacheHandle segment_cache;
RETURN_IF_ERROR(SegmentLoader::instance()->load_segments(
std::static_pointer_cast<BetaRowset>(rs), &segment_cache));
for (const auto& column : rs->tablet_schema()->columns()) {
if (!column->is_variant_type() || column->unique_id() < 0) {
continue;
}
if (!should_check_variant_path_stats(*column)) {
continue;
}
for (const auto& segment : segment_cache.get_segments()) {
std::shared_ptr<ColumnReader> column_reader;
OlapReaderStatistics stats;
RETURN_IF_ERROR(
segment->get_column_reader(column->unique_id(), &column_reader, &stats));
if (!column_reader) {
continue;
}
CHECK(column_reader->get_meta_type() == FieldType::OLAP_FIELD_TYPE_VARIANT);
auto* variant_column_reader =
assert_cast<segment_v2::VariantColumnReader*>(column_reader.get());
// load external meta before getting stats
RETURN_IF_ERROR(variant_column_reader->load_external_meta_once());
const auto* source_stats = variant_column_reader->get_stats();
CHECK(source_stats);
// agg path -> stats
for (const auto& [path, size] : source_stats->sparse_column_non_null_size) {
(*uid_to_path_stats)[column->unique_id()][path] += size;
}
for (const auto& [path, size] : source_stats->subcolumns_non_null_size) {
(*uid_to_path_stats)[column->unique_id()][path] += size;
}
}
}
return Status::OK();
}
Status VariantCompactionUtil::aggregate_variant_extended_info(
const RowsetSharedPtr& rs,
std::unordered_map<int32_t, VariantExtendedInfo>* uid_to_variant_extended_info) {
SegmentCacheHandle segment_cache;
RETURN_IF_ERROR(SegmentLoader::instance()->load_segments(
std::static_pointer_cast<BetaRowset>(rs), &segment_cache));
for (const auto& column : rs->tablet_schema()->columns()) {
if (!column->is_variant_type()) {
continue;
}
auto& extended_info = (*uid_to_variant_extended_info)[column->unique_id()];
if (column->variant_enable_nested_group()) {
extended_info.has_nested_group = true;
}
for (const auto& segment : segment_cache.get_segments()) {
std::shared_ptr<ColumnReader> column_reader;
OlapReaderStatistics stats;
RETURN_IF_ERROR(
segment->get_column_reader(column->unique_id(), &column_reader, &stats));
if (!column_reader) {
continue;
}
CHECK(column_reader->get_meta_type() == FieldType::OLAP_FIELD_TYPE_VARIANT);
auto* variant_column_reader =
assert_cast<segment_v2::VariantColumnReader*>(column_reader.get());
// load external meta before getting stats
RETURN_IF_ERROR(variant_column_reader->load_external_meta_once());
const auto* source_stats = variant_column_reader->get_stats();
CHECK(source_stats);
if (!column->variant_enable_nested_group()) {
// NG roots still need type metadata for regular subpaths such as `v.owner`,
// but their compaction schema should not be driven by flat path stats.
for (const auto& [path, size] : source_stats->sparse_column_non_null_size) {
extended_info.path_to_none_null_values[path] += size;
extended_info.sparse_paths.emplace(path);
}
for (const auto& [path, size] : source_stats->subcolumns_non_null_size) {
extended_info.path_to_none_null_values[path] += size;
}
}
//2. agg path -> schema
variant_column_reader->get_subcolumns_types(&extended_info.path_to_data_types);
// 3. extract typed paths
variant_column_reader->get_typed_paths(&extended_info.typed_paths);
// 4. extract nested paths
if (!column->variant_enable_nested_group()) {
variant_column_reader->get_nested_paths(&extended_info.nested_paths);
}
}
}
return Status::OK();
}
// get the subpaths and sparse paths for the variant column
void VariantCompactionUtil::get_subpaths(int32_t max_subcolumns_count,
const PathToNoneNullValues& stats,
TabletSchema::PathsSetInfo& paths_set_info) {
// max_subcolumns_count is 0 means no limit
if (max_subcolumns_count > 0 && stats.size() > max_subcolumns_count) {
std::vector<std::pair<size_t, std::string_view>> paths_with_sizes;
paths_with_sizes.reserve(stats.size());
for (const auto& [path, size] : stats) {
paths_with_sizes.emplace_back(size, path);
}
std::sort(paths_with_sizes.begin(), paths_with_sizes.end(), std::greater());
// Select top N paths as subcolumns, remaining paths as sparse columns
for (const auto& [size, path] : paths_with_sizes) {
if (paths_set_info.sub_path_set.size() < max_subcolumns_count) {
paths_set_info.sub_path_set.emplace(path);
} else {
paths_set_info.sparse_path_set.emplace(path);
}
}
LOG(INFO) << "subpaths " << paths_set_info.sub_path_set.size() << " sparse paths "
<< paths_set_info.sparse_path_set.size() << " variant max subcolumns count "
<< max_subcolumns_count << " stats size " << paths_with_sizes.size();
} else {
// Apply all paths as subcolumns
for (const auto& [path, _] : stats) {
paths_set_info.sub_path_set.emplace(path);
}
}
}
Status VariantCompactionUtil::check_path_stats(const std::vector<RowsetSharedPtr>& intputs,
RowsetSharedPtr output, BaseTabletSPtr tablet) {
if (output->tablet_schema()->num_variant_columns() == 0) {
return Status::OK();
}
for (const auto& rowset : intputs) {
for (const auto& column : rowset->tablet_schema()->columns()) {
if (column->is_variant_type() && !should_check_variant_path_stats(*column)) {
return Status::OK();
}
}
}
// check no extended schema in input rowsets
for (const auto& rowset : intputs) {
for (const auto& column : rowset->tablet_schema()->columns()) {
if (column->is_extracted_column()) {
return Status::OK();
}
}
}
#ifndef BE_TEST
// check no extended schema in output rowset
for (const auto& column : output->tablet_schema()->columns()) {
if (column->is_extracted_column()) {
const auto& name = column->name();
if (name.find("." + DOC_VALUE_COLUMN_PATH + ".") != std::string::npos ||
name.find("." + SPARSE_COLUMN_PATH + ".") != std::string::npos ||
name.ends_with("." + SPARSE_COLUMN_PATH)) {
continue;
}
return Status::InternalError("Unexpected extracted column {} in output rowset",
column->name());
}
}
#endif
// only check path stats for dup_keys since the rows may be merged in other models
if (tablet->keys_type() != KeysType::DUP_KEYS) {
return Status::OK();
}
// if there is a delete predicate in the input rowsets, we skip the path stats check
for (auto& rowset : intputs) {
if (rowset->rowset_meta()->has_delete_predicate()) {
return Status::OK();
}
}
for (const auto& column : output->tablet_schema()->columns()) {
if (column->is_variant_type() && !should_check_variant_path_stats(*column)) {
return Status::OK();
}
}
std::unordered_map<int32_t, PathToNoneNullValues> original_uid_to_path_stats;
for (const auto& rs : intputs) {
RETURN_IF_ERROR(aggregate_path_to_stats(rs, &original_uid_to_path_stats));
}
std::unordered_map<int32_t, PathToNoneNullValues> output_uid_to_path_stats;
RETURN_IF_ERROR(aggregate_path_to_stats(output, &output_uid_to_path_stats));
for (const auto& [uid, stats] : output_uid_to_path_stats) {
if (output->tablet_schema()->column_by_uid(uid).is_variant_type() &&
output->tablet_schema()->column_by_uid(uid).variant_enable_doc_mode()) {
continue;
}
if (original_uid_to_path_stats.find(uid) == original_uid_to_path_stats.end()) {
return Status::InternalError("Path stats not found for uid {}, tablet_id {}", uid,
tablet->tablet_id());
}
// In input rowsets, some rowsets may have statistics values exceeding the maximum limit,
// which leads to inaccurate statistics
if (stats.size() > output->tablet_schema()
->column_by_uid(uid)
.variant_max_sparse_column_statistics_size()) {
// When there is only one segment, we can ensure that the size of each path in output stats is accurate
if (output->num_segments() == 1) {
for (const auto& [path, size] : stats) {
if (original_uid_to_path_stats.at(uid).find(path) ==
original_uid_to_path_stats.at(uid).end()) {
continue;
}
if (original_uid_to_path_stats.at(uid).at(path) > size) {
return Status::InternalError(
"Path stats not smaller for uid {} with path `{}`, input size {}, "
"output "
"size {}, "
"tablet_id {}",
uid, path, original_uid_to_path_stats.at(uid).at(path), size,
tablet->tablet_id());
}
}
}
}
// in this case, input stats is accurate, so we check the stats size and stats value
else {
for (const auto& [path, size] : stats) {
if (original_uid_to_path_stats.at(uid).find(path) ==
original_uid_to_path_stats.at(uid).end()) {
return Status::InternalError(
"Path stats not found for uid {}, path {}, tablet_id {}", uid, path,
tablet->tablet_id());
}
if (original_uid_to_path_stats.at(uid).at(path) != size) {
return Status::InternalError(
"Path stats not match for uid {} with path `{}`, input size {}, output "
"size {}, "
"tablet_id {}",
uid, path, original_uid_to_path_stats.at(uid).at(path), size,
tablet->tablet_id());
}
}
}
}
return Status::OK();
}
Status VariantCompactionUtil::get_compaction_typed_columns(
const TabletSchemaSPtr& target, const std::unordered_set<std::string>& typed_paths,
const TabletColumnPtr parent_column, TabletSchemaSPtr& output_schema,
TabletSchema::PathsSetInfo& paths_set_info) {
if (parent_column->variant_enable_typed_paths_to_sparse()) {
return Status::OK();
}
for (const auto& path : typed_paths) {
TabletSchema::SubColumnInfo sub_column_info;
if (generate_sub_column_info(*target, parent_column->unique_id(), path, &sub_column_info)) {
inherit_column_attributes(*parent_column, sub_column_info.column);
output_schema->append_column(sub_column_info.column);
paths_set_info.typed_path_set.insert({path, std::move(sub_column_info)});
VLOG_DEBUG << "append typed column " << path;
} else {
return Status::InternalError("Failed to generate sub column info for path {}", path);
}
}
return Status::OK();
}
Status VariantCompactionUtil::get_compaction_nested_columns(
const std::unordered_set<PathInData, PathInData::Hash>& nested_paths,
const PathToDataTypes& path_to_data_types, const TabletColumnPtr parent_column,
TabletSchemaSPtr& output_schema, TabletSchema::PathsSetInfo& paths_set_info) {
const auto& parent_indexes = output_schema->inverted_indexs(parent_column->unique_id());
for (const auto& path : nested_paths) {
const auto& find_data_types = path_to_data_types.find(path);
if (find_data_types == path_to_data_types.end() || find_data_types->second.empty()) {
return Status::InternalError("Nested path {} has no data type", path.get_path());
}
DataTypePtr data_type;
get_least_supertype_jsonb(find_data_types->second, &data_type);
const std::string& column_name = parent_column->name_lower_case() + "." + path.get_path();
PathInDataBuilder full_path_builder;
auto full_path = full_path_builder.append(parent_column->name_lower_case(), false)
.append(path.get_parts(), false)
.build();
TabletColumn nested_column =
get_column_by_type(data_type, column_name,
ExtraInfo {.unique_id = -1,
.parent_unique_id = parent_column->unique_id(),
.path_info = full_path});
inherit_column_attributes(*parent_column, nested_column);
TabletIndexes sub_column_indexes;
inherit_index(parent_indexes, sub_column_indexes, nested_column);
paths_set_info.subcolumn_indexes.emplace(path.get_path(), std::move(sub_column_indexes));
output_schema->append_column(nested_column);
VLOG_DEBUG << "append nested column " << path.get_path();
}
return Status::OK();
}
void VariantCompactionUtil::get_compaction_subcolumns_from_subpaths(
TabletSchema::PathsSetInfo& paths_set_info, const TabletColumnPtr parent_column,
const TabletSchemaSPtr& target, const PathToDataTypes& path_to_data_types,
const std::unordered_set<std::string>& sparse_paths, TabletSchemaSPtr& output_schema) {
auto& path_set = paths_set_info.sub_path_set;
std::vector<StringRef> sorted_subpaths(path_set.begin(), path_set.end());
std::sort(sorted_subpaths.begin(), sorted_subpaths.end());
const auto& parent_indexes = target->inverted_indexs(parent_column->unique_id());
// append subcolumns
for (const auto& subpath : sorted_subpaths) {
auto column_name = parent_column->name_lower_case() + "." + subpath.to_string();
auto column_path = PathInData(column_name);
const auto& find_data_types = path_to_data_types.find(PathInData(subpath));
// some cases: the subcolumn type is variant
// 1. this path has no data type in segments
// 2. this path is in sparse paths
// 3. the sparse paths are too much
TabletSchema::SubColumnInfo sub_column_info;
if (parent_column->variant_enable_typed_paths_to_sparse() &&
generate_sub_column_info(*target, parent_column->unique_id(), std::string(subpath),
&sub_column_info)) {
inherit_column_attributes(*parent_column, sub_column_info.column);
output_schema->append_column(sub_column_info.column);
paths_set_info.subcolumn_indexes.emplace(subpath, std::move(sub_column_info.indexes));
VLOG_DEBUG << "append typed column " << subpath;
} else if (find_data_types == path_to_data_types.end() || find_data_types->second.empty() ||
sparse_paths.find(std::string(subpath)) != sparse_paths.end() ||
sparse_paths.size() >=
parent_column->variant_max_sparse_column_statistics_size()) {
TabletColumn subcolumn;
subcolumn.set_name(column_name);
subcolumn.set_type(FieldType::OLAP_FIELD_TYPE_VARIANT);
subcolumn.set_parent_unique_id(parent_column->unique_id());
subcolumn.set_path_info(column_path);
subcolumn.set_aggregation_method(parent_column->aggregation());
subcolumn.set_variant_max_subcolumns_count(
parent_column->variant_max_subcolumns_count());
subcolumn.set_variant_enable_doc_mode(parent_column->variant_enable_doc_mode());
subcolumn.set_is_nullable(true);
output_schema->append_column(subcolumn);
VLOG_DEBUG << "append sub column " << subpath << " data type "
<< "VARIANT";
}
// normal case: the subcolumn type can be calculated from the data types in segments
else {
DataTypePtr data_type;
get_least_supertype_jsonb(find_data_types->second, &data_type);
TabletColumn sub_column =
get_column_by_type(data_type, column_name,
ExtraInfo {.unique_id = -1,
.parent_unique_id = parent_column->unique_id(),
.path_info = column_path});
inherit_column_attributes(*parent_column, sub_column);
TabletIndexes sub_column_indexes;
inherit_index(parent_indexes, sub_column_indexes, sub_column);
paths_set_info.subcolumn_indexes.emplace(subpath, std::move(sub_column_indexes));
output_schema->append_column(sub_column);
VLOG_DEBUG << "append sub column " << subpath << " data type " << data_type->get_name();
}
}
}
void VariantCompactionUtil::get_compaction_subcolumns_from_data_types(
TabletSchema::PathsSetInfo& paths_set_info, const TabletColumnPtr parent_column,
const TabletSchemaSPtr& target, const PathToDataTypes& path_to_data_types,
TabletSchemaSPtr& output_schema) {
const auto& parent_indexes = target->inverted_indexs(parent_column->unique_id());
for (const auto& [path, data_types] : path_to_data_types) {
if (data_types.empty() || path.empty() || path.has_nested_part()) {
continue;
}
DataTypePtr data_type;
get_least_supertype_jsonb(data_types, &data_type);
auto column_name = parent_column->name_lower_case() + "." + path.get_path();
auto column_path = PathInData(column_name, path.get_is_typed());
TabletColumn sub_column =
get_column_by_type(data_type, column_name,
ExtraInfo {.unique_id = -1,
.parent_unique_id = parent_column->unique_id(),
.path_info = column_path});
inherit_column_attributes(*parent_column, sub_column);
TabletIndexes sub_column_indexes;
inherit_index(parent_indexes, sub_column_indexes, sub_column);
if (path.get_is_typed()) {
TabletSchema::SubColumnInfo sub_column_info {.column = sub_column,
.indexes = std::move(sub_column_indexes)};
paths_set_info.typed_path_set.emplace(path.get_path(), std::move(sub_column_info));
} else {
paths_set_info.subcolumn_indexes.emplace(path.get_path(),
std::move(sub_column_indexes));
}
output_schema->append_column(sub_column);
VLOG_DEBUG << "append sub column " << path.get_path() << " data type "
<< data_type->get_name();
}
}
// Build the temporary schema for compaction
// 1. aggregate path stats and data types from all rowsets
// 2. append typed columns and nested columns to the output schema
// 3. sort the subpaths and sparse paths for each unique id
// 4. append the subpaths and sparse paths to the output schema
// 5. set the path set info for each unique id
// 6. return the output schema
Status VariantCompactionUtil::get_extended_compaction_schema(
const std::vector<RowsetSharedPtr>& rowsets, TabletSchemaSPtr& target) {
std::unordered_map<int32_t, VariantExtendedInfo> uid_to_variant_extended_info;
const bool has_extendable_variant =
std::ranges::any_of(target->columns(), [](const TabletColumnPtr& column) {
return column->is_variant_type() && should_check_variant_path_stats(*column);
});
if (has_extendable_variant) {
// collect path stats from all rowsets and segments
for (const auto& rs : rowsets) {
RETURN_IF_ERROR(aggregate_variant_extended_info(rs, &uid_to_variant_extended_info));
}
}
// build the output schema
TabletSchemaSPtr output_schema = std::make_shared<TabletSchema>();
output_schema->shawdow_copy_without_columns(*target);
std::unordered_map<int32_t, TabletSchema::PathsSetInfo> uid_to_paths_set_info;
const auto ng_root_uids = collect_ng_compaction_root_uids(target, uid_to_variant_extended_info);
const auto uid_to_existing_ng_subcolumns =
collect_existing_ng_compaction_subcolumns(target, ng_root_uids);
for (const TabletColumnPtr& column : target->columns()) {
if (!column->is_extracted_column()) {
output_schema->append_column(*column);
}
if (!column->is_variant_type()) {
continue;
}
VLOG_DEBUG << "column " << column->name() << " unique id " << column->unique_id();
const auto info_it = uid_to_variant_extended_info.find(column->unique_id());
const VariantExtendedInfo empty_extended_info;
const VariantExtendedInfo& extended_info = info_it == uid_to_variant_extended_info.end()
? empty_extended_info
: info_it->second;
auto& paths_set_info = uid_to_paths_set_info[column->unique_id()];
if (ng_root_uids.contains(column->unique_id())) {
const auto plan = build_nested_group_compaction_materialization_plan(
uid_to_existing_ng_subcolumns.contains(column->unique_id())
? uid_to_existing_ng_subcolumns.at(column->unique_id())
: std::vector<TabletColumnPtr> {},
extended_info);
append_nested_group_compaction_columns(target, column, plan, output_schema,
paths_set_info);
LOG(INFO) << "Variant column uid=" << column->unique_id()
<< " keeps nested-group root with regular extracted columns in compaction "
"schema";
continue;
}
if (!should_check_variant_path_stats(*column)) {
VLOG_DEBUG << "skip extended schema compaction for variant uid=" << column->unique_id()
<< " because the column disables variant path stats";
continue;
}
if (extended_info.has_nested_group) {
LOG(INFO) << "Variant column uid=" << column->unique_id()
<< " has nested group, keep original column in compaction schema";
continue;
}
if (column->variant_enable_doc_mode()) {
const int bucket_num = std::max(1, column->variant_doc_hash_shard_count());
for (int b = 0; b < bucket_num; ++b) {
TabletColumn doc_value_bucket_column = create_doc_value_column(*column, b);
doc_value_bucket_column.set_type(FieldType::OLAP_FIELD_TYPE_VARIANT);
doc_value_bucket_column.set_is_nullable(false);
doc_value_bucket_column.set_variant_enable_doc_mode(true);
output_schema->append_column(doc_value_bucket_column);
}
continue;
}
// 1. append typed columns
RETURN_IF_ERROR(get_compaction_typed_columns(target, extended_info.typed_paths, column,
output_schema, paths_set_info));
// 2. append nested columns
RETURN_IF_ERROR(get_compaction_nested_columns(extended_info.nested_paths,
extended_info.path_to_data_types, column,
output_schema, paths_set_info));
// 3. get the subpaths
get_subpaths(column->variant_max_subcolumns_count(), extended_info.path_to_none_null_values,
paths_set_info);
// 4. append subcolumns
if (column->variant_max_subcolumns_count() > 0 || !column->get_sub_columns().empty()) {
get_compaction_subcolumns_from_subpaths(paths_set_info, column, target,
extended_info.path_to_data_types,
extended_info.sparse_paths, output_schema);
}
// variant_max_subcolumns_count == 0 and no typed paths materialized
// it means that all subcolumns are materialized, may be from old data
else {
get_compaction_subcolumns_from_data_types(paths_set_info, column, target,
extended_info.path_to_data_types,
output_schema);
}
// append sparse column(s)
// If variant uses bucketized sparse columns, append one sparse bucket column per bucket.
// Otherwise, append the single sparse column.
int bucket_num = std::max(1, column->variant_sparse_hash_shard_count());
if (bucket_num > 1) {
for (int b = 0; b < bucket_num; ++b) {
TabletColumn sparse_bucket_column = create_sparse_shard_column(*column, b);
output_schema->append_column(sparse_bucket_column);
}
} else {
TabletColumn sparse_column = create_sparse_column(*column);
output_schema->append_column(sparse_column);
}
}
target = output_schema;
// used to merge & filter path to sparse column during reading in compaction
target->set_path_set_info(std::move(uid_to_paths_set_info));
VLOG_DEBUG << "dump schema " << target->dump_full_schema();
return Status::OK();
}
// Calculate statistics about variant data paths from the encoded sparse column
void VariantCompactionUtil::calculate_variant_stats(const IColumn& encoded_sparse_column,
segment_v2::VariantStatisticsPB* stats,
size_t max_sparse_column_statistics_size,
size_t row_pos, size_t num_rows) {
// Cast input column to ColumnMap type since sparse column is stored as a map
const auto& map_column = assert_cast<const ColumnMap&>(encoded_sparse_column);
// Get the keys column which contains the paths as strings
const auto& sparse_data_paths =
assert_cast<const ColumnString*>(map_column.get_keys_ptr().get());
const auto& serialized_sparse_column_offsets =
assert_cast<const ColumnArray::Offsets64&>(map_column.get_offsets());
auto& count_map = *stats->mutable_sparse_column_non_null_size();
// Iterate through all paths in the sparse column
for (size_t i = row_pos; i != row_pos + num_rows; ++i) {
size_t offset = serialized_sparse_column_offsets[i - 1];
size_t end = serialized_sparse_column_offsets[i];
for (size_t j = offset; j != end; ++j) {
auto path = sparse_data_paths->get_data_at(j);
const auto& sparse_path = path.to_string();
// If path already exists in statistics, increment its count
if (auto it = count_map.find(sparse_path); it != count_map.end()) {
++it->second;
}
// If path doesn't exist and we haven't hit the max statistics size limit,
// add it with count 1
else if (count_map.size() < max_sparse_column_statistics_size) {
count_map.emplace(sparse_path, 1);
}
}
}
if (stats->sparse_column_non_null_size().size() > max_sparse_column_statistics_size) {
throw doris::Exception(
ErrorCode::INTERNAL_ERROR,
"Sparse column non null size: {} is greater than max statistics size: {}",
stats->sparse_column_non_null_size().size(), max_sparse_column_statistics_size);
}
}
/// Calculates number of dimensions in array field.
/// Returns 0 for scalar fields.
class FieldVisitorToNumberOfDimensions : public StaticVisitor<size_t> {
public:
FieldVisitorToNumberOfDimensions() = default;
template <PrimitiveType T>
size_t apply(const typename PrimitiveTypeTraits<T>::CppType& x) {
if constexpr (T == TYPE_ARRAY) {
const size_t size = x.size();
size_t dimensions = 0;
for (size_t i = 0; i < size; ++i) {
size_t element_dimensions = apply_visitor(*this, x[i]);
dimensions = std::max(dimensions, element_dimensions);
}
return 1 + dimensions;
} else {
return 0;
}
}
};
// Visitor that allows to get type of scalar field
// but exclude fields contain complex field.This is a faster version
// for FieldVisitorToScalarType which does not support complex field.
class SimpleFieldVisitorToScalarType : public StaticVisitor<size_t> {
public:
template <PrimitiveType T>
size_t apply(const typename PrimitiveTypeTraits<T>::CppType& x) {
if constexpr (T == TYPE_ARRAY) {
throw doris::Exception(ErrorCode::INVALID_ARGUMENT, "Array type is not supported");
} else if constexpr (T == TYPE_NULL) {
have_nulls = true;
return 1;
} else {
type = T;
return 1;
}
}
void get_scalar_type(PrimitiveType* data_type) const { *data_type = type; }
bool contain_nulls() const { return have_nulls; }
bool need_convert_field() const { return false; }
private:
PrimitiveType type = PrimitiveType::INVALID_TYPE;
bool have_nulls = false;
};
/// Visitor that allows to get type of scalar field
/// or least common type of scalars in array.
/// More optimized version of FieldToDataType.
class FieldVisitorToScalarType : public StaticVisitor<size_t> {
public:
template <PrimitiveType T>
size_t apply(const typename PrimitiveTypeTraits<T>::CppType& x) {
if constexpr (T == TYPE_ARRAY) {
size_t size = x.size();
for (size_t i = 0; i < size; ++i) {
apply_visitor(*this, x[i]);
}
return 0;
} else if constexpr (T == TYPE_NULL) {
have_nulls = true;
return 0;
} else {
field_types.insert(T);
type_indexes.insert(T);
return 0;
}
}
void get_scalar_type(PrimitiveType* type) const {
if (type_indexes.size() == 1) {
// Most cases will have only one type
*type = *type_indexes.begin();
return;
}
DataTypePtr data_type;
get_least_supertype_jsonb(type_indexes, &data_type);
*type = data_type->get_primitive_type();
}
bool contain_nulls() const { return have_nulls; }
bool need_convert_field() const { return field_types.size() > 1; }
private:
phmap::flat_hash_set<PrimitiveType> type_indexes;
phmap::flat_hash_set<PrimitiveType> field_types;
bool have_nulls = false;
};
template <typename Visitor>
void get_field_info_impl(const Field& field, FieldInfo* info) {
Visitor to_scalar_type_visitor;
apply_visitor(to_scalar_type_visitor, field);
PrimitiveType type_id;
to_scalar_type_visitor.get_scalar_type(&type_id);
// array item's dimension may missmatch, eg. [1, 2, [1, 2, 3]]
*info = {type_id, to_scalar_type_visitor.contain_nulls(),
to_scalar_type_visitor.need_convert_field(),
apply_visitor(FieldVisitorToNumberOfDimensions(), field)};
}
void get_field_info(const Field& field, FieldInfo* info) {
if (field.is_complex_field()) {
get_field_info_impl<FieldVisitorToScalarType>(field, info);
} else {
get_field_info_impl<SimpleFieldVisitorToScalarType>(field, info);
}
}
bool generate_sub_column_info(const TabletSchema& schema, int32_t col_unique_id,
const std::string& path,
TabletSchema::SubColumnInfo* sub_column_info) {
const auto& parent_column = schema.column_by_uid(col_unique_id);
std::function<void(const TabletColumn&, TabletColumn*)> generate_result_column =
[&](const TabletColumn& from_column, TabletColumn* to_column) {
to_column->set_name(parent_column.name_lower_case() + "." + path);
to_column->set_type(from_column.type());
to_column->set_parent_unique_id(parent_column.unique_id());
bool is_typed = !parent_column.variant_enable_typed_paths_to_sparse();
to_column->set_path_info(
PathInData(parent_column.name_lower_case() + "." + path, is_typed));
to_column->set_aggregation_method(parent_column.aggregation());
to_column->set_is_nullable(true);
to_column->set_parent_unique_id(parent_column.unique_id());
if (from_column.is_decimal()) {
to_column->set_precision(from_column.precision());
}
to_column->set_frac(from_column.frac());
if (from_column.is_array_type()) {
TabletColumn nested_column;
generate_result_column(*from_column.get_sub_columns()[0], &nested_column);
to_column->add_sub_column(nested_column);
}
};
auto generate_index = [&](const std::string& pattern) {
// 1. find subcolumn's index
if (const auto& indexes = schema.inverted_index_by_field_pattern(col_unique_id, pattern);
!indexes.empty()) {
for (const auto& index : indexes) {
auto index_ptr = std::make_shared<TabletIndex>(*index);
index_ptr->set_escaped_escaped_index_suffix_path(
sub_column_info->column.path_info_ptr()->get_path());
sub_column_info->indexes.emplace_back(std::move(index_ptr));
}
}
// 2. find parent column's index
else if (const auto parent_index = schema.inverted_indexs(col_unique_id);
!parent_index.empty()) {
inherit_index(parent_index, sub_column_info->indexes, sub_column_info->column);
} else {
sub_column_info->indexes.clear();
}
};
const auto& sub_columns = parent_column.get_sub_columns();
for (const auto& sub_column : sub_columns) {
const char* pattern = sub_column->name().c_str();
switch (sub_column->pattern_type()) {
case PatternTypePB::MATCH_NAME: {
if (strcmp(pattern, path.c_str()) == 0) {
generate_result_column(*sub_column, &sub_column_info->column);
generate_index(sub_column->name());
return true;
}
break;
}
case PatternTypePB::MATCH_NAME_GLOB: {
if (glob_match_re2(pattern, path)) {
generate_result_column(*sub_column, &sub_column_info->column);
generate_index(sub_column->name());
return true;
}
break;
}
default:
break;
}
}
return false;
}
TabletSchemaSPtr VariantCompactionUtil::calculate_variant_extended_schema(
const std::vector<RowsetSharedPtr>& rowsets, const TabletSchemaSPtr& base_schema) {
if (rowsets.empty()) {
return nullptr;
}
std::vector<TabletSchemaSPtr> schemas;
for (const auto& rs : rowsets) {
if (rs->num_segments() == 0) {
continue;
}
const auto& tablet_schema = rs->tablet_schema();
SegmentCacheHandle segment_cache;
auto st = SegmentLoader::instance()->load_segments(std::static_pointer_cast<BetaRowset>(rs),
&segment_cache);
if (!st.ok()) {
return base_schema;
}
for (const auto& segment : segment_cache.get_segments()) {
TabletSchemaSPtr schema = tablet_schema->copy_without_variant_extracted_columns();
for (const auto& column : tablet_schema->columns()) {
if (!column->is_variant_type()) {
continue;
}
std::shared_ptr<ColumnReader> column_reader;
OlapReaderStatistics stats;
st = segment->get_column_reader(column->unique_id(), &column_reader, &stats);
if (!st.ok()) {
LOG(WARNING) << "Failed to get column reader for column: " << column->name()
<< " error: " << st.to_string();
continue;
}
if (!column_reader) {
continue;
}
CHECK(column_reader->get_meta_type() == FieldType::OLAP_FIELD_TYPE_VARIANT);
auto* variant_column_reader =
assert_cast<segment_v2::VariantColumnReader*>(column_reader.get());
// load external meta before getting subcolumn meta info
st = variant_column_reader->load_external_meta_once();
if (!st.ok()) {
LOG(WARNING) << "Failed to load external meta for column: " << column->name()
<< " error: " << st.to_string();
continue;
}
const auto* subcolumn_meta_info = variant_column_reader->get_subcolumns_meta_info();
for (const auto& entry : *subcolumn_meta_info) {
if (entry->path.empty()) {
continue;
}
const std::string& column_name =
column->name_lower_case() + "." + entry->path.get_path();
const DataTypePtr& data_type = entry->data.file_column_type;
PathInDataBuilder full_path_builder;
auto full_path = full_path_builder.append(column->name_lower_case(), false)
.append(entry->path.get_parts(), false)
.build();
TabletColumn subcolumn =
get_column_by_type(data_type, column_name,
ExtraInfo {.unique_id = -1,
.parent_unique_id = column->unique_id(),
.path_info = full_path});
schema->append_column(subcolumn);
}
}
schemas.emplace_back(schema);
}
}
TabletSchemaSPtr least_common_schema;
auto st = get_least_common_schema(schemas, base_schema, least_common_schema, false);
if (!st.ok()) {
return base_schema;
}
return least_common_schema;
}
bool inherit_index(const std::vector<const TabletIndex*>& parent_indexes,
TabletIndexes& subcolumns_indexes, FieldType column_type,
const std::string& suffix_path, bool is_array_nested_type) {
if (parent_indexes.empty()) {
return false;
}
subcolumns_indexes.clear();
// bkd index or array index only need to inherit one index
if (field_is_numeric_type(column_type) ||
(is_array_nested_type &&
(field_is_numeric_type(column_type) || field_is_slice_type(column_type)))) {
auto index_ptr = std::make_shared<TabletIndex>(*parent_indexes[0]);
index_ptr->set_escaped_escaped_index_suffix_path(suffix_path);
// no need parse for bkd index or array index
index_ptr->remove_parser_and_analyzer();
subcolumns_indexes.emplace_back(std::move(index_ptr));
return true;
}
// string type need to inherit all indexes
else if (field_is_slice_type(column_type) && !is_array_nested_type) {
for (const auto& index : parent_indexes) {
auto index_ptr = std::make_shared<TabletIndex>(*index);
index_ptr->set_escaped_escaped_index_suffix_path(suffix_path);
subcolumns_indexes.emplace_back(std::move(index_ptr));
}
return true;
}
return false;
}
bool inherit_index(const std::vector<const TabletIndex*>& parent_indexes,
TabletIndexes& subcolumns_indexes, const TabletColumn& column) {
if (!column.is_extracted_column()) {
return false;
}
if (column.is_array_type()) {
if (column.get_sub_columns().empty()) {
return false;
}
const TabletColumn* nested = column.get_sub_columns()[0].get();
while (nested != nullptr && nested->is_array_type()) {
if (nested->get_sub_columns().empty()) {
return false;
}
nested = nested->get_sub_columns()[0].get();
}
if (nested == nullptr) {
return false;
}
return inherit_index(parent_indexes, subcolumns_indexes, nested->type(),
column.path_info_ptr()->get_path(), true);
}
return inherit_index(parent_indexes, subcolumns_indexes, column.type(),
column.path_info_ptr()->get_path());
}
bool inherit_index(const std::vector<const TabletIndex*>& parent_indexes,
TabletIndexes& subcolumns_indexes, const ColumnMetaPB& column_pb) {
if (!column_pb.has_column_path_info()) {
return false;
}
if (column_pb.type() == (int)FieldType::OLAP_FIELD_TYPE_ARRAY) {
if (column_pb.children_columns_size() == 0) {
return false;
}
const ColumnMetaPB* nested = &column_pb.children_columns(0);
while (nested != nullptr && nested->type() == (int)FieldType::OLAP_FIELD_TYPE_ARRAY) {
if (nested->children_columns_size() == 0) {
return false;
}
nested = &nested->children_columns(0);
}
if (nested == nullptr) {
return false;
}
return inherit_index(parent_indexes, subcolumns_indexes, (FieldType)nested->type(),
column_pb.column_path_info().path(), true);
}
return inherit_index(parent_indexes, subcolumns_indexes, (FieldType)column_pb.type(),
column_pb.column_path_info().path());
}
// ============ Implementation from parse2column.cpp ============
/** Pool for objects that cannot be used from different threads simultaneously.
* Allows to create an object for each thread.
* Pool has unbounded size and objects are not destroyed before destruction of pool.
*
* Use it in cases when thread local storage is not appropriate
* (when maximum number of simultaneously used objects is less
* than number of running/sleeping threads, that has ever used object,
* and creation/destruction of objects is expensive).
*/
template <typename T>
class SimpleObjectPool {
protected:
/// Hold all available objects in stack.
std::mutex mutex;
std::stack<std::unique_ptr<T>> stack;
/// Specialized deleter for std::unique_ptr.
/// Returns underlying pointer back to stack thus reclaiming its ownership.
struct Deleter {
SimpleObjectPool<T>* parent;
Deleter(SimpleObjectPool<T>* parent_ = nullptr) : parent {parent_} {} /// NOLINT
void operator()(T* owning_ptr) const {
std::lock_guard lock {parent->mutex};
parent->stack.emplace(owning_ptr);
}
};
public:
using Pointer = std::unique_ptr<T, Deleter>;
/// Extracts and returns a pointer from the stack if it's not empty,
/// creates a new one by calling provided f() otherwise.
template <typename Factory>
Pointer get(Factory&& f) {
std::unique_lock lock(mutex);
if (stack.empty()) {
return {f(), this};
}
auto object = stack.top().release();
stack.pop();
return std::unique_ptr<T, Deleter>(object, Deleter(this));
}
/// Like get(), but creates object using default constructor.
Pointer getDefault() {
return get([] { return new T; });
}
};
SimpleObjectPool<JsonParser> parsers_pool;
using Node = typename ColumnVariant::Subcolumns::Node;
static inline void append_binary_bytes(ColumnString::Chars& chars, const void* data, size_t size) {
const auto old_size = chars.size();
chars.resize(old_size + size);
memcpy(chars.data() + old_size, reinterpret_cast<const char*>(data), size);
}
static inline void append_binary_type(ColumnString::Chars& chars, FieldType type) {
const uint8_t t = static_cast<uint8_t>(type);
append_binary_bytes(chars, &t, sizeof(uint8_t));
}
static inline void append_binary_sizet(ColumnString::Chars& chars, size_t v) {
append_binary_bytes(chars, &v, sizeof(size_t));
}
static void append_field_to_binary_chars(const Field& field, ColumnString::Chars& chars) {
switch (field.get_type()) {
case PrimitiveType::TYPE_NULL: {
append_binary_type(chars, FieldType::OLAP_FIELD_TYPE_NONE);
return;
}
case PrimitiveType::TYPE_BOOLEAN: {
append_binary_type(chars,
TabletColumn::get_field_type_by_type(PrimitiveType::TYPE_BOOLEAN));
const auto v = static_cast<UInt8>(field.get<PrimitiveType::TYPE_BOOLEAN>());
append_binary_bytes(chars, &v, sizeof(UInt8));
return;
}
case PrimitiveType::TYPE_BIGINT: {
append_binary_type(chars, TabletColumn::get_field_type_by_type(PrimitiveType::TYPE_BIGINT));
const auto v = field.get<PrimitiveType::TYPE_BIGINT>();
append_binary_bytes(chars, &v, sizeof(Int64));
return;
}
case PrimitiveType::TYPE_LARGEINT: {
append_binary_type(chars,
TabletColumn::get_field_type_by_type(PrimitiveType::TYPE_LARGEINT));
const auto v = field.get<PrimitiveType::TYPE_LARGEINT>();
append_binary_bytes(chars, &v, sizeof(int128_t));
return;
}
case PrimitiveType::TYPE_DOUBLE: {
append_binary_type(chars, TabletColumn::get_field_type_by_type(PrimitiveType::TYPE_DOUBLE));
const auto v = field.get<PrimitiveType::TYPE_DOUBLE>();
append_binary_bytes(chars, &v, sizeof(Float64));
return;
}
case PrimitiveType::TYPE_STRING: {
append_binary_type(chars, FieldType::OLAP_FIELD_TYPE_STRING);
const auto& v = field.get<PrimitiveType::TYPE_STRING>();
append_binary_sizet(chars, v.size());
append_binary_bytes(chars, v.data(), v.size());
return;
}
case PrimitiveType::TYPE_JSONB: {
append_binary_type(chars, FieldType::OLAP_FIELD_TYPE_JSONB);
const auto& v = field.get<PrimitiveType::TYPE_JSONB>();
append_binary_sizet(chars, v.get_size());
append_binary_bytes(chars, v.get_value(), v.get_size());
return;
}
case PrimitiveType::TYPE_ARRAY: {
append_binary_type(chars, FieldType::OLAP_FIELD_TYPE_ARRAY);
const auto& a = field.get<PrimitiveType::TYPE_ARRAY>();
append_binary_sizet(chars, a.size());
for (const auto& elem : a) {
append_field_to_binary_chars(elem, chars);
}
return;
}
default:
throw doris::Exception(ErrorCode::INVALID_ARGUMENT, "Unsupported field type {}",
field.get_type());
}
}
template <typename ParserImpl>
void parse_json_to_variant_impl(IColumn& column, const char* src, size_t length,
JSONDataParser<ParserImpl>* parser, const ParseConfig& config) {
auto& column_variant = assert_cast<ColumnVariant&>(column);
std::optional<ParseResult> result;
/// Treat empty string as an empty object
/// for better CAST from String to Object.
if (length > 0) {
result = parser->parse(src, length, config);
} else {
result = ParseResult {};
}
if (!result) {
VLOG_DEBUG << "failed to parse " << std::string_view(src, length) << ", length= " << length;
if (config::variant_throw_exeception_on_invalid_json) {
throw doris::Exception(ErrorCode::INVALID_ARGUMENT, "Failed to parse object {}",
std::string_view(src, length));
}
// Treat as string
PathInData root_path;
Field field = Field::create_field<TYPE_STRING>(String(src, length));
result = ParseResult {{root_path}, {field}};
}
auto& [paths, values] = *result;
assert(paths.size() == values.size());
size_t old_num_rows = column_variant.rows();
if (config.deprecated_enable_flatten_nested) {
// here we should check the paths in variant and paths in result,
// if two paths which same prefix have different structure, we should throw an exception
std::vector<PathInData> check_paths;
for (const auto& entry : column_variant.get_subcolumns()) {
check_paths.push_back(entry->path);
}
check_paths.insert(check_paths.end(), paths.begin(), paths.end());
THROW_IF_ERROR(check_variant_has_no_ambiguous_paths(check_paths));
}
auto [doc_value_data_paths, doc_value_data_values] =
column_variant.get_doc_value_data_paths_and_values();
auto& doc_value_data_offsets = column_variant.serialized_doc_value_column_offsets();
auto flush_defaults = [](ColumnVariant::Subcolumn* subcolumn) {
const auto num_defaults = subcolumn->cur_num_of_defaults();
if (num_defaults > 0) {
subcolumn->insert_many_defaults(num_defaults);
subcolumn->reset_current_num_of_defaults();
}
};
auto get_or_create_subcolumn = [&](const PathInData& path, size_t index_hint,
const FieldInfo& field_info) -> ColumnVariant::Subcolumn* {
if (column_variant.get_subcolumn(path, index_hint) == nullptr) {
if (path.has_nested_part()) {
column_variant.add_nested_subcolumn(path, field_info, old_num_rows);
} else {
column_variant.add_sub_column(path, old_num_rows);
}
}
auto* subcolumn = column_variant.get_subcolumn(path, index_hint);
if (!subcolumn) {
throw doris::Exception(ErrorCode::INVALID_ARGUMENT, "Failed to find sub column {}",
path.get_path());
}
return subcolumn;
};
auto insert_into_subcolumn = [&](size_t i,
bool check_size_mismatch) -> ColumnVariant::Subcolumn* {
FieldInfo field_info;
get_field_info(values[i], &field_info);
if (field_info.scalar_type_id == PrimitiveType::INVALID_TYPE) {
return nullptr;
}
auto* subcolumn = get_or_create_subcolumn(paths[i], i, field_info);
flush_defaults(subcolumn);
if (check_size_mismatch && subcolumn->size() != old_num_rows) {
throw doris::Exception(ErrorCode::INVALID_ARGUMENT,
"subcolumn {} size missmatched, may contains duplicated entry",
paths[i].get_path());
}
subcolumn->insert(std::move(values[i]), std::move(field_info));
return subcolumn;
};
switch (config.parse_to) {
case ParseConfig::ParseTo::OnlySubcolumns:
for (size_t i = 0; i < paths.size(); ++i) {
insert_into_subcolumn(i, true);
}
break;
case ParseConfig::ParseTo::OnlyDocValueColumn: {
CHECK(column_variant.enable_doc_mode()) << "OnlyDocValueColumn requires doc mode enabled";
std::vector<size_t> doc_item_indexes;
doc_item_indexes.reserve(paths.size());
phmap::flat_hash_set<StringRef, StringRefHash> seen_paths;
seen_paths.reserve(paths.size());
for (size_t i = 0; i < paths.size(); ++i) {
FieldInfo field_info;
get_field_info(values[i], &field_info);
if (paths[i].empty()) {
auto* subcolumn = column_variant.get_subcolumn(paths[i]);
DCHECK(subcolumn != nullptr);
flush_defaults(subcolumn);
subcolumn->insert(std::move(values[i]), std::move(field_info));
continue;
}
if (field_info.scalar_type_id == PrimitiveType::INVALID_TYPE ||
values[i].get_type() == PrimitiveType::TYPE_NULL) {
continue;
}
const auto& path_str = paths[i].get_path();
StringRef path_ref {path_str.data(), path_str.size()};
if (UNLIKELY(!seen_paths.emplace(path_ref).second)) {
throw doris::Exception(ErrorCode::INVALID_ARGUMENT,
"may contains duplicated entry : {}",
std::string_view(path_str));
}
doc_item_indexes.push_back(i);
}
std::sort(doc_item_indexes.begin(), doc_item_indexes.end(),
[&](size_t l, size_t r) { return paths[l].get_path() < paths[r].get_path(); });
for (const auto idx : doc_item_indexes) {
const auto& path_str = paths[idx].get_path();
doc_value_data_paths->insert_data(path_str.data(), path_str.size());
auto& chars = doc_value_data_values->get_chars();
append_field_to_binary_chars(values[idx], chars);
doc_value_data_values->get_offsets().push_back(chars.size());
}
} break;
}
doc_value_data_offsets.push_back(doc_value_data_paths->size());
// /// Insert default values to missed subcolumns.
const auto& subcolumns = column_variant.get_subcolumns();
for (const auto& entry : subcolumns) {
if (entry->data.size() == old_num_rows) {
// Handle nested paths differently from simple paths
if (entry->path.has_nested_part()) {
// Try to insert default from nested, if failed, insert regular default
bool success = UNLIKELY(column_variant.try_insert_default_from_nested(entry));
if (!success) {
entry->data.insert_default();
}
} else {
// For non-nested paths, increment default counter
entry->data.increment_default_counter();
}
}
}
column_variant.incr_num_rows();
auto sparse_column = column_variant.get_sparse_column();
if (sparse_column->size() == old_num_rows) {
sparse_column->assume_mutable()->insert_default();
}
#ifndef NDEBUG
column_variant.check_consistency();
#endif
}
// exposed interfaces
void parse_json_to_variant(IColumn& column, const StringRef& json, JsonParser* parser,
const ParseConfig& config) {
if (parser) {
return parse_json_to_variant_impl(column, json.data, json.size, parser, config);
} else {
auto pool_parser = parsers_pool.get([] { return new JsonParser(); });
return parse_json_to_variant_impl(column, json.data, json.size, pool_parser.get(), config);
}
}
void parse_json_to_variant(IColumn& column, const ColumnString& raw_json_column,
const ParseConfig& config) {
auto parser = parsers_pool.get([] { return new JsonParser(); });
for (size_t i = 0; i < raw_json_column.size(); ++i) {
StringRef raw_json = raw_json_column.get_data_at(i);
parse_json_to_variant_impl(column, raw_json.data, raw_json.size, parser.get(), config);
}
column.finalize();
}
// parse the doc snapshot column to subcolumns
void materialize_docs_to_subcolumns(ColumnVariant& column_variant) {
auto subcolumns = materialize_docs_to_subcolumns_map(column_variant);
for (auto& entry : subcolumns) {
entry.second.finalize();
if (!column_variant.add_sub_column(PathInData(entry.first),
IColumn::mutate(entry.second.get_finalized_column_ptr()),
entry.second.get_least_common_type())) {
throw doris::Exception(ErrorCode::INTERNAL_ERROR,
"Failed to add subcolumn {}, which is from doc snapshot column",
entry.first);
}
}
column_variant.finalize();
}
// ============ Implementation from variant_util.cpp ============
phmap::flat_hash_map<std::string_view, ColumnVariant::Subcolumn> materialize_docs_to_subcolumns_map(
const ColumnVariant& variant, size_t expected_unique_paths) {
constexpr size_t kInitialPathReserve = 8192;
phmap::flat_hash_map<std::string_view, ColumnVariant::Subcolumn> subcolumns;
const auto [column_key, column_value] = variant.get_doc_value_data_paths_and_values();
const auto& column_offsets = variant.serialized_doc_value_column_offsets();
const size_t num_rows = column_offsets.size();
DCHECK_EQ(num_rows, variant.size()) << "doc snapshot offsets size mismatch with variant rows";
subcolumns.reserve(expected_unique_paths != 0
? expected_unique_paths
: std::min<size_t>(column_key->size(), kInitialPathReserve));
for (size_t row = 0; row < num_rows; ++row) {
const size_t start = column_offsets[row - 1];
const size_t end = column_offsets[row];
for (size_t i = start; i < end; ++i) {
const auto& key = column_key->get_data_at(i);
const std::string_view path_sv(key.data, key.size);
auto [it, inserted] =
subcolumns.try_emplace(path_sv, ColumnVariant::Subcolumn {0, true, false});
auto& subcolumn = it->second;
if (inserted) {
subcolumn.insert_many_defaults(row);
} else if (subcolumn.size() != row) {
subcolumn.insert_many_defaults(row - subcolumn.size());
}
subcolumn.deserialize_from_binary_column(column_value, i);
}
}
for (auto& [path, subcolumn] : subcolumns) {
if (subcolumn.size() != num_rows) {
subcolumn.insert_many_defaults(num_rows - subcolumn.size());
}
}
return subcolumns;
}
Status _parse_and_materialize_variant_columns(Block& block,
const std::vector<uint32_t>& variant_pos,
const std::vector<ParseConfig>& configs) {
for (size_t i = 0; i < variant_pos.size(); ++i) {
auto column_ref = block.get_by_position(variant_pos[i]).column;
bool is_nullable = column_ref->is_nullable();
MutableColumnPtr var_column = column_ref->assume_mutable();
if (is_nullable) {
const auto& nullable = assert_cast<const ColumnNullable&>(*column_ref);
var_column = nullable.get_nested_column_ptr()->assume_mutable();
}
auto& var = assert_cast<ColumnVariant&>(*var_column);
var_column->finalize();
MutableColumnPtr variant_column;
if (!var.is_scalar_variant()) {
// already parsed
continue;
}
VLOG_DEBUG << "parse scalar variant column: " << var.get_root_type()->get_name();
ColumnPtr scalar_root_column;
if (var.get_root_type()->get_primitive_type() == TYPE_JSONB) {
// TODO more efficient way to parse jsonb type, currently we just convert jsonb to
// json str and parse them into variant
RETURN_IF_ERROR(cast_column({var.get_root(), var.get_root_type(), ""},
var.get_root()->is_nullable()
? make_nullable(std::make_shared<DataTypeString>())
: std::make_shared<DataTypeString>(),
&scalar_root_column));
if (scalar_root_column->is_nullable()) {
scalar_root_column = assert_cast<const ColumnNullable*>(scalar_root_column.get())
->get_nested_column_ptr();
}
} else {
const auto& root = *var.get_root();
scalar_root_column =
root.is_nullable()
? assert_cast<const ColumnNullable&>(root).get_nested_column_ptr()
: var.get_root();
}
if (scalar_root_column->is_column_string()) {
variant_column = ColumnVariant::create(0, var.enable_doc_mode());
parse_json_to_variant(*variant_column.get(),
assert_cast<const ColumnString&>(*scalar_root_column),
configs[i]);
} else {
// Root maybe other types rather than string like ColumnVariant(Int32).
// In this case, we should finlize the root and cast to JSON type
auto expected_root_type =
make_nullable(std::make_shared<ColumnVariant::MostCommonType>());
var.ensure_root_node_type(expected_root_type);
variant_column = var.assume_mutable();
}
// Wrap variant with nullmap if it is nullable
ColumnPtr result = variant_column->get_ptr();
if (is_nullable) {
const auto& null_map =
assert_cast<const ColumnNullable&>(*column_ref).get_null_map_column_ptr();
result = ColumnNullable::create(result, null_map);
}
block.get_by_position(variant_pos[i]).column = result;
}
return Status::OK();
}
Status parse_and_materialize_variant_columns(Block& block, const std::vector<uint32_t>& variant_pos,
const std::vector<ParseConfig>& configs) {
RETURN_IF_CATCH_EXCEPTION(
{ return _parse_and_materialize_variant_columns(block, variant_pos, configs); });
}
Status parse_and_materialize_variant_columns(Block& block, const TabletSchema& tablet_schema,
const std::vector<uint32_t>& column_pos) {
std::vector<uint32_t> variant_column_pos;
std::vector<uint32_t> variant_schema_pos;
variant_column_pos.reserve(column_pos.size());
variant_schema_pos.reserve(column_pos.size());
for (size_t block_pos = 0; block_pos < column_pos.size(); ++block_pos) {
const uint32_t schema_pos = column_pos[block_pos];
const auto& column = tablet_schema.column(schema_pos);
if (column.is_variant_type()) {
variant_column_pos.push_back(schema_pos);
variant_schema_pos.push_back(schema_pos);
}
}
if (variant_column_pos.empty()) {
return Status::OK();
}
std::vector<ParseConfig> configs(variant_column_pos.size());
for (size_t i = 0; i < variant_column_pos.size(); ++i) {
// Deprecated legacy flatten-nested switch. Distinct from variant_enable_nested_group.
configs[i].deprecated_enable_flatten_nested =
tablet_schema.deprecated_variant_flatten_nested();
const auto& column = tablet_schema.column(variant_schema_pos[i]);
if (!column.is_variant_type()) {
return Status::InternalError("column is not variant type, column name: {}",
column.name());
}
// if doc mode is not enabled, no need to parse to doc value column
if (!column.variant_enable_doc_mode()) {
configs[i].parse_to = ParseConfig::ParseTo::OnlySubcolumns;
continue;
}
configs[i].parse_to = ParseConfig::ParseTo::OnlyDocValueColumn;
}
RETURN_IF_ERROR(parse_and_materialize_variant_columns(block, variant_column_pos, configs));
return Status::OK();
}
} // namespace doris::variant_util
Reference in new issue View Git Blame Copy Permalink