Google Git
Sign in
apache/doris/HEAD/./be/src/format_v2/table_reader.cpp
blob: 4beaf8c9ff555040817829ad278583bd807fcd0f [file]
// 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 "format_v2/table_reader.h"
#include <gen_cpp/ExternalTableSchema_types.h>
#include <gen_cpp/PlanNodes_types.h>
#include <gen_cpp/Types_types.h>
#include <algorithm>
#include <memory>
#include <ranges>
#include <set>
#include <sstream>
#include <utility>
#include <vector>
#include "common/cast_set.h"
#include "common/status.h"
#include "core/assert_cast.h"
#include "core/data_type/data_type_array.h"
#include "core/data_type/data_type_factory.hpp"
#include "core/data_type/data_type_map.h"
#include "core/data_type/data_type_struct.h"
#include "core/data_type/primitive_type.h"
#include "exprs/vexpr_context.h"
#include "exprs/vslot_ref.h"
#include "format/table/deletion_vector_reader.h"
#include "format/table/iceberg_delete_file_reader_helper.h"
#include "format/table/iceberg_scan_semantics.h"
#include "format/table/paimon_reader.h"
#include "format_v2/column_mapper.h"
#include "format_v2/delimited_text/csv_reader.h"
#include "format_v2/delimited_text/text_reader.h"
#include "format_v2/json/json_reader.h"
#include "format_v2/native/native_reader.h"
#include "format_v2/orc/orc_reader.h"
#include "format_v2/parquet/parquet_reader.h"
#include "runtime/file_scan_profile.h"
#include "storage/segment/condition_cache.h"
#include "util/debug_points.h"
#include "util/string_util.h"
namespace doris::format {
namespace {
template <typename T, typename Formatter>
std::string join_table_reader_debug_strings(const std::vector<T>& values, Formatter formatter) {
std::ostringstream out;
out << "[";
for (size_t i = 0; i < values.size(); ++i) {
if (i > 0) {
out << ", ";
}
out << formatter(values[i]);
}
out << "]";
return out.str();
}
std::string file_format_to_string(FileFormat format) {
switch (format) {
case FileFormat::PARQUET:
return "PARQUET";
case FileFormat::ORC:
return "ORC";
case FileFormat::CSV:
return "CSV";
case FileFormat::JSON:
return "JSON";
case FileFormat::TEXT:
return "TEXT";
case FileFormat::JNI:
return "JNI";
case FileFormat::NATIVE:
return "NATIVE";
case FileFormat::ARROW:
return "ARROW";
}
return "UNKNOWN";
}
std::string push_down_agg_to_string(TPushAggOp::type op) {
switch (op) {
case TPushAggOp::NONE:
return "NONE";
case TPushAggOp::COUNT:
return "COUNT";
case TPushAggOp::MINMAX:
return "MINMAX";
case TPushAggOp::MIX:
return "MIX";
case TPushAggOp::COUNT_ON_INDEX:
return "COUNT_ON_INDEX";
}
return "UNKNOWN";
}
std::string current_file_debug_string(const std::unique_ptr<ScanTask>& task) {
if (task == nullptr || task->data_file == nullptr) {
return "null";
}
const auto& file = *task->data_file;
std::ostringstream out;
out << "FileDescription{path=" << file.path << ", file_size=" << file.file_size
<< ", range_start_offset=" << file.range_start_offset << ", range_size=" << file.range_size
<< ", mtime=" << file.mtime << ", fs_name=" << file.fs_name
<< ", is_immutable=" << file.is_immutable
<< ", file_cache_admission=" << file.file_cache_admission << "}";
return out.str();
}
std::string partition_values_debug_string(const std::map<std::string, Field>& partition_values) {
std::ostringstream out;
out << "{";
size_t idx = 0;
for (const auto& [key, _] : partition_values) {
if (idx++ > 0) {
out << ", ";
}
out << key;
}
out << "}";
return out.str();
}
const schema::external::TField* get_field_ptr(const schema::external::TFieldPtr& field_ptr) {
if (!field_ptr.__isset.field_ptr || field_ptr.field_ptr == nullptr) {
return nullptr;
}
return field_ptr.field_ptr.get();
}
ColumnDefinition build_schema_identity_from_external_field(const schema::external::TField& field) {
ColumnDefinition identity;
if (field.__isset.id) {
identity.identifier = Field::create_field<TYPE_INT>(field.id);
}
identity.name = field.__isset.name ? field.name : "";
identity.name_mapping =
field.__isset.name_mapping ? field.name_mapping : std::vector<std::string> {};
identity.has_name_mapping =
field.__isset.name_mapping_is_authoritative && field.name_mapping_is_authoritative;
if (!field.__isset.nestedField) {
return identity;
}
if (field.nestedField.__isset.struct_field && field.nestedField.struct_field.__isset.fields) {
for (const auto& child_ptr : field.nestedField.struct_field.fields) {
if (const auto* child = get_field_ptr(child_ptr); child != nullptr) {
identity.children.push_back(build_schema_identity_from_external_field(*child));
}
}
} else if (field.nestedField.__isset.array_field &&
field.nestedField.array_field.__isset.item_field) {
if (const auto* child = get_field_ptr(field.nestedField.array_field.item_field);
child != nullptr) {
identity.children.push_back(build_schema_identity_from_external_field(*child));
identity.children.back().name = "element";
}
} else if (field.nestedField.__isset.map_field) {
if (field.nestedField.map_field.__isset.key_field) {
if (const auto* child = get_field_ptr(field.nestedField.map_field.key_field);
child != nullptr) {
identity.children.push_back(build_schema_identity_from_external_field(*child));
identity.children.back().name = "key";
}
}
if (field.nestedField.map_field.__isset.value_field) {
if (const auto* child = get_field_ptr(field.nestedField.map_field.value_field);
child != nullptr) {
identity.children.push_back(build_schema_identity_from_external_field(*child));
identity.children.back().name = "value";
}
}
}
return identity;
}
const ColumnDefinition* find_identity_child(const ColumnDefinition& projected_child,
const ColumnDefinition& identity_parent) {
const auto child_it = std::ranges::find_if(
identity_parent.children, [&](const ColumnDefinition& identity_child) {
if (projected_child.has_identifier_field_id() &&
identity_child.has_identifier_field_id()) {
return projected_child.get_identifier_field_id() ==
identity_child.get_identifier_field_id();
}
if (to_lower(projected_child.name) == to_lower(identity_child.name)) {
return true;
}
return std::ranges::any_of(
identity_child.name_mapping, [&](const std::string& alias) {
return to_lower(projected_child.name) == to_lower(alias);
});
});
return child_it == identity_parent.children.end() ? nullptr : &*child_it;
}
void attach_full_schema_identity(ColumnDefinition* projected, const ColumnDefinition& identity) {
DORIS_CHECK(projected != nullptr);
// Access-path children control materialization, but wrapper discovery needs sibling IDs that
// were pruned from that projection. Keep the complete identity tree on a separate channel.
projected->identity_children = identity.children;
for (auto& projected_child : projected->children) {
if (const auto* identity_child = find_identity_child(projected_child, identity);
identity_child != nullptr) {
attach_full_schema_identity(&projected_child, *identity_child);
}
}
}
void clear_initial_default_metadata(ColumnDefinition* column) {
DORIS_CHECK(column != nullptr);
column->initial_default_value.reset();
column->initial_default_value_is_base64 = false;
for (auto& child : column->children) {
clear_initial_default_metadata(&child);
}
}
bool external_field_matches_name(const schema::external::TField& field, const std::string& name) {
if (field.__isset.name && to_lower(field.name) == to_lower(name)) {
return true;
}
return field.__isset.name_mapping &&
std::ranges::any_of(field.name_mapping, [&](const std::string& alias) {
return to_lower(alias) == to_lower(name);
});
}
DataTypePtr find_struct_child_type_by_external_field(const DataTypeStruct& struct_type,
const schema::external::TField& field) {
for (size_t field_idx = 0; field_idx < struct_type.get_elements().size(); ++field_idx) {
if (external_field_matches_name(field, struct_type.get_element_name(field_idx))) {
return struct_type.get_element(field_idx);
}
}
return nullptr;
}
DataTypePtr restore_current_primitive_type(const schema::external::TField& field,
DataTypePtr fallback_type) {
if (!field.__isset.type) {
return fallback_type;
}
const auto primitive_type = thrift_to_type(field.type.type);
if (is_complex_type(primitive_type)) {
return fallback_type;
}
// The delete file can expose an older physical type, but initial defaults belong to the
// current table field. Restore that type from FE before parsing the default and let the table
// reader apply the normal promotion cast to the delete-key type.
return DataTypeFactory::instance().create_data_type(
primitive_type, false, field.type.__isset.precision ? field.type.precision : 0,
field.type.__isset.scale ? field.type.scale : 0,
field.type.__isset.len ? field.type.len : -1);
}
ColumnDefinition build_schema_column_from_external_field(const schema::external::TField& field,
DataTypePtr type) {
type = restore_current_primitive_type(field, std::move(type));
ColumnDefinition column {
.identifier = field.__isset.id ? Field::create_field<TYPE_INT>(field.id) : Field {},
.name = field.__isset.name ? field.name : "",
.name_mapping =
field.__isset.name_mapping ? field.name_mapping : std::vector<std::string> {},
.has_name_mapping = field.__isset.name_mapping_is_authoritative &&
field.name_mapping_is_authoritative,
.type = std::move(type),
.children = {},
.default_expr = nullptr,
.initial_default_value = field.__isset.initial_default_value
? std::make_optional(field.initial_default_value)
: std::nullopt,
.initial_default_value_is_base64 = field.__isset.initial_default_value_is_base64 &&
field.initial_default_value_is_base64,
.is_partition_key = false,
};
if (column.type == nullptr || !field.__isset.nestedField) {
return column;
}
const auto nested_type = remove_nullable(column.type);
switch (nested_type->get_primitive_type()) {
case TYPE_STRUCT: {
if (!field.nestedField.__isset.struct_field ||
!field.nestedField.struct_field.__isset.fields) {
return column;
}
const auto& struct_type = assert_cast<const DataTypeStruct&>(*nested_type);
for (const auto& child_ptr : field.nestedField.struct_field.fields) {
const auto* child_field = get_field_ptr(child_ptr);
if (child_field == nullptr || !child_field->__isset.name) {
continue;
}
auto child_type = find_struct_child_type_by_external_field(struct_type, *child_field);
if (child_type == nullptr) {
continue;
}
column.children.push_back(
build_schema_column_from_external_field(*child_field, child_type));
}
break;
}
case TYPE_ARRAY: {
if (!field.nestedField.__isset.array_field ||
!field.nestedField.array_field.__isset.item_field) {
return column;
}
const auto* item_field = get_field_ptr(field.nestedField.array_field.item_field);
if (item_field == nullptr) {
return column;
}
const auto& array_type = assert_cast<const DataTypeArray&>(*nested_type);
auto child =
build_schema_column_from_external_field(*item_field, array_type.get_nested_type());
child.name = "element";
if (child.has_identifier_name()) {
child.identifier = Field::create_field<TYPE_STRING>(child.name);
}
column.children.push_back(std::move(child));
break;
}
case TYPE_MAP: {
if (!field.nestedField.__isset.map_field ||
!field.nestedField.map_field.__isset.key_field ||
!field.nestedField.map_field.__isset.value_field) {
return column;
}
const auto& map_type = assert_cast<const DataTypeMap&>(*nested_type);
const auto* key_field = get_field_ptr(field.nestedField.map_field.key_field);
if (key_field != nullptr) {
auto child =
build_schema_column_from_external_field(*key_field, map_type.get_key_type());
child.name = "key";
if (child.has_identifier_name()) {
child.identifier = Field::create_field<TYPE_STRING>(child.name);
}
column.children.push_back(std::move(child));
}
const auto* value_field = get_field_ptr(field.nestedField.map_field.value_field);
if (value_field != nullptr) {
auto child = build_schema_column_from_external_field(*value_field,
map_type.get_value_type());
child.name = "value";
if (child.has_identifier_name()) {
child.identifier = Field::create_field<TYPE_STRING>(child.name);
}
column.children.push_back(std::move(child));
}
break;
}
default:
break;
}
return column;
}
const schema::external::TField* find_external_root_field(const TFileScanRangeParams* params,
const ColumnDefinition& column) {
if (params == nullptr || !params->__isset.history_schema_info ||
params->history_schema_info.empty()) {
return nullptr;
}
const auto* schema = &params->history_schema_info.front();
if (params->__isset.current_schema_id) {
for (const auto& candidate_schema : params->history_schema_info) {
if (candidate_schema.__isset.schema_id &&
candidate_schema.schema_id == params->current_schema_id) {
schema = &candidate_schema;
break;
}
}
}
if (!schema->__isset.root_field || !schema->root_field.__isset.fields) {
return nullptr;
}
if (!supports_iceberg_scan_semantics_v1(params)) {
// Old BEs used one ordered current-name/alias pass. Preserve that result for old-FE plans
// until the explicit scan-semantics marker makes exact-name precedence cluster-wide.
for (const auto& field_ptr : schema->root_field.fields) {
const auto* field = get_field_ptr(field_ptr);
if (field != nullptr && external_field_matches_name(*field, column.name)) {
return field;
}
}
return nullptr;
}
// A reused name identifies the newly added field, not an older sibling that retained that
// spelling as an alias. Exhaust exact current names before consulting historical aliases.
for (const auto& field_ptr : schema->root_field.fields) {
const auto* field = get_field_ptr(field_ptr);
if (field != nullptr && field->__isset.name &&
to_lower(field->name) == to_lower(column.name)) {
return field;
}
}
for (const auto& field_ptr : schema->root_field.fields) {
const auto* field = get_field_ptr(field_ptr);
if (field != nullptr && field->__isset.name_mapping &&
std::ranges::any_of(field->name_mapping, [&](const std::string& alias) {
return to_lower(alias) == to_lower(column.name);
})) {
return field;
}
}
return nullptr;
}
std::string expr_context_debug_string(const VExprContextSPtr& context) {
if (context == nullptr) {
return "null";
}
const auto root = context->root();
if (root == nullptr) {
return "VExprContext{root=null}";
}
std::ostringstream out;
out << "VExprContext{root_name=" << root->expr_name() << ", root_debug=" << root->debug_string()
<< "}";
return out.str();
}
std::string table_filter_debug_string(const TableFilter& filter) {
std::ostringstream out;
out << "TableFilter{conjunct=" << expr_context_debug_string(filter.conjunct)
<< ", global_indices="
<< join_table_reader_debug_strings(
filter.global_indices,
[](GlobalIndex global_index) { return std::to_string(global_index.value()); })
<< "}";
return out.str();
}
bool contains_runtime_filter(const VExprContextSPtrs& conjuncts) {
return std::ranges::any_of(conjuncts, [](const auto& conjunct) {
return conjunct != nullptr && conjunct->root() != nullptr &&
conjunct->root()->is_rf_wrapper();
});
}
void collect_global_indices(const VExprSPtr& expr, std::set<GlobalIndex>* global_indices) {
if (expr == nullptr) {
return;
}
if (expr->is_rf_wrapper()) {
// RuntimeFilterExpr wraps a real predicate expression but its own thrift node can still
// look like SLOT_REF. Collect indices from the wrapped predicate; do not cast the wrapper
// itself to VSlotRef.
collect_global_indices(expr->get_impl(), global_indices);
return;
}
if (expr->is_slot_ref()) {
const auto* slot_ref = assert_cast<const VSlotRef*>(expr.get());
DORIS_CHECK(slot_ref->column_id() >= 0);
global_indices->insert(GlobalIndex(cast_set<size_t>(slot_ref->column_id())));
}
for (const auto& child : expr->children()) {
collect_global_indices(child, global_indices);
}
}
Status build_table_filters_from_conjunct(const VExprContextSPtr& conjunct, RuntimeState* state,
std::vector<TableFilter>* table_filters) {
if (conjunct == nullptr) {
return Status::OK();
}
std::set<GlobalIndex> global_indices;
collect_global_indices(conjunct->root(), &global_indices);
if (!global_indices.empty()) {
TableFilter table_filter;
VExprSPtr filter_root;
RETURN_IF_ERROR(clone_table_expr_tree(conjunct->root(), &filter_root));
table_filter.conjunct = VExprContext::create_shared(std::move(filter_root));
for (const auto global_index : global_indices) {
table_filter.global_indices.push_back(global_index);
}
table_filters->push_back(std::move(table_filter));
}
return Status::OK();
}
Status parse_deletion_vector(const char* buf, size_t buffer_size, DeleteFileDesc::Format format,
DeletionVector* deletion_vector) {
DORIS_CHECK(buf != nullptr);
DORIS_CHECK(deletion_vector != nullptr);
DORIS_CHECK(format == DeleteFileDesc::Format::PAIMON ||
format == DeleteFileDesc::Format::ICEBERG);
if (format == DeleteFileDesc::Format::PAIMON) {
RETURN_IF_ERROR(decode_paimon_deletion_vector_buffer(buf, buffer_size, deletion_vector));
return Status::OK();
}
return decode_iceberg_deletion_vector_buffer(buf, buffer_size, deletion_vector);
}
} // namespace
std::shared_ptr<io::FileSystemProperties> create_system_properties(
const TFileScanRangeParams* scan_params) {
auto system_properties = std::make_shared<io::FileSystemProperties>();
if (scan_params == nullptr || !scan_params->__isset.file_type) {
system_properties->system_type = TFileType::FILE_LOCAL;
return system_properties;
}
system_properties->system_type = scan_params->file_type;
system_properties->properties = scan_params->properties;
system_properties->hdfs_params = scan_params->hdfs_params;
if (scan_params->__isset.broker_addresses) {
system_properties->broker_addresses.assign(scan_params->broker_addresses.begin(),
scan_params->broker_addresses.end());
}
return system_properties;
}
std::string TableReader::debug_string() const {
std::ostringstream out;
out << "TableReader{format=" << file_format_to_string(_format)
<< ", push_down_agg_type=" << push_down_agg_to_string(_push_down_agg_type)
<< ", aggregate_pushdown_tried=" << _aggregate_pushdown_tried
<< ", has_current_reader=" << (_data_reader.reader != nullptr)
<< ", has_current_task=" << (_current_task != nullptr)
<< ", current_file=" << current_file_debug_string(_current_task)
<< ", has_delete_rows=" << (_delete_rows != nullptr)
<< ", delete_row_count=" << (_delete_rows == nullptr ? 0 : _delete_rows->size())
<< ", has_deletion_vector=" << (_deletion_vector != nullptr)
<< ", deletion_vector_cardinality="
<< (_deletion_vector == nullptr ? 0 : _deletion_vector->cardinality())
<< ", has_system_properties=" << (_system_properties != nullptr) << ", system_type="
<< (_system_properties == nullptr ? static_cast<int>(TFileType::FILE_LOCAL)
: static_cast<int>(_system_properties->system_type))
<< ", has_scan_params=" << (_scan_params != nullptr)
<< ", has_io_ctx=" << (_io_ctx != nullptr)
<< ", has_runtime_state=" << (_runtime_state != nullptr)
<< ", has_scanner_profile=" << (_scanner_profile != nullptr)
<< ", mapper_options=" << _mapper_options.debug_string() << ", projected_columns="
<< join_table_reader_debug_strings(
_projected_columns,
[](const ColumnDefinition& column) { return column.debug_string(); })
<< ", partition_values=" << partition_values_debug_string(_partition_values)
<< ", table_filters="
<< join_table_reader_debug_strings(
_table_filters,
[](const TableFilter& filter) { return table_filter_debug_string(filter); })
<< ", conjunct_count=" << _conjuncts.size() << ", conjuncts="
<< join_table_reader_debug_strings(_conjuncts,
[](const VExprContextSPtr& conjunct) {
return expr_context_debug_string(conjunct);
})
<< ", file_schema="
<< join_table_reader_debug_strings(
_data_reader.file_schema,
[](const ColumnDefinition& field) { return field.debug_string(); })
<< ", file_block_layout="
<< join_table_reader_debug_strings(
_data_reader.file_block_layout,
[](const FileBlockColumn& column) {
std::ostringstream column_out;
column_out << "FileBlockColumn{file_column_id=" << column.file_column_id
<< ", name=" << column.name << ", type="
<< (column.type == nullptr ? "null" : column.type->get_name())
<< "}";
return column_out.str();
})
<< ", block_template_columns=" << _data_reader.block_template.columns()
<< ", column_mapper="
<< (_data_reader.column_mapper == nullptr ? "null"
: _data_reader.column_mapper->debug_string())
<< "}";
return out.str();
}
Status TableReader::annotate_projected_column(const TFileScanSlotInfo& slot_info,
ProjectedColumnBuildContext* context,
ColumnDefinition* column) const {
(void)slot_info;
DORIS_CHECK(context != nullptr);
DORIS_CHECK(column != nullptr);
context->schema_column.reset();
const auto* schema_field = find_external_root_field(context->scan_params, *column);
if (schema_field == nullptr) {
return Status::OK();
}
context->schema_column = build_schema_column_from_external_field(*schema_field, column->type);
const bool use_current_semantics = supports_iceberg_scan_semantics_v1(context->scan_params);
if (!use_current_semantics) {
// IDs and encoded defaults predate the result-changing semantics. Strip only the new
// default channel so an old-FE plan keeps the same generic root/nested values on every BE.
clear_initial_default_metadata(&*context->schema_column);
}
column->identifier = context->schema_column->identifier;
column->name_mapping = context->schema_column->name_mapping;
column->has_name_mapping = context->schema_column->has_name_mapping;
// Projected roots already carry a generic FE default expression, but Iceberg binary defaults
// need the raw Base64 marker so missing-file materialization can decode rather than copy text.
column->initial_default_value = context->schema_column->initial_default_value;
column->initial_default_value_is_base64 =
context->schema_column->initial_default_value_is_base64;
return Status::OK();
}
std::optional<ColumnDefinition> TableReader::_find_current_table_column_by_field_id(
int32_t field_id, DataTypePtr type) const {
if (_scan_params == nullptr || !_scan_params->__isset.history_schema_info ||
_scan_params->history_schema_info.empty()) {
return std::nullopt;
}
const auto* schema = &_scan_params->history_schema_info.front();
if (_scan_params->__isset.current_schema_id) {
for (const auto& candidate_schema : _scan_params->history_schema_info) {
if (candidate_schema.__isset.schema_id &&
candidate_schema.schema_id == _scan_params->current_schema_id) {
schema = &candidate_schema;
break;
}
}
}
if (!schema->__isset.root_field || !schema->root_field.__isset.fields) {
return std::nullopt;
}
for (const auto& field_ptr : schema->root_field.fields) {
const auto* field = get_field_ptr(field_ptr);
if (field != nullptr && field->__isset.id && field->id == field_id) {
return build_schema_column_from_external_field(*field, std::move(type));
}
}
return std::nullopt;
}
Status TableReader::init(TableReadOptions&& options) {
_scanner_profile = options.scanner_profile;
if (_scanner_profile != nullptr) {
const auto hierarchy = file_scan_profile::ensure_hierarchy(_scanner_profile);
static const char* table_profile = file_scan_profile::TABLE_READER;
static const char* file_reader_profile = file_scan_profile::FILE_READER;
_profile.total_timer = hierarchy.table_reader;
_profile.file_reader_total_timer = hierarchy.file_reader;
_profile.init_timer =
ADD_CHILD_TIMER_WITH_LEVEL(_scanner_profile, "InitTime", table_profile, 1);
_profile.num_delete_files = ADD_CHILD_COUNTER_WITH_LEVEL(_scanner_profile, "NumDeleteFiles",
TUnit::UNIT, table_profile, 1);
_profile.num_delete_rows = ADD_CHILD_COUNTER_WITH_LEVEL(_scanner_profile, "NumDeleteRows",
TUnit::UNIT, table_profile, 1);
_profile.parse_delete_file_time = ADD_CHILD_TIMER_WITH_LEVEL(
_scanner_profile, "ParseDeleteFileTime", table_profile, 1);
_profile.decoded_dv_cache_hit_count =
ADD_CHILD_COUNTER_WITH_LEVEL(_scanner_profile, "DeletionVectorDecodedCacheHitCount",
TUnit::UNIT, table_profile, 1);
_profile.decoded_dv_cache_miss_count = ADD_CHILD_COUNTER_WITH_LEVEL(
_scanner_profile, "DeletionVectorDecodedCacheMissCount", TUnit::UNIT, table_profile,
1);
_profile.dv_file_cache_hit_count = ADD_CHILD_COUNTER_WITH_LEVEL(
_scanner_profile, "DeletionVectorFileCacheHitCount", TUnit::UNIT, table_profile, 1);
_profile.dv_file_cache_miss_count =
ADD_CHILD_COUNTER_WITH_LEVEL(_scanner_profile, "DeletionVectorFileCacheMissCount",
TUnit::UNIT, table_profile, 1);
_profile.dv_file_cache_peer_read_count = ADD_CHILD_COUNTER_WITH_LEVEL(
_scanner_profile, "DeletionVectorFileCachePeerReadCount", TUnit::UNIT,
table_profile, 1);
_profile.exec_timer =
ADD_CHILD_TIMER_WITH_LEVEL(_scanner_profile, "GetBlockTime", table_profile, 1);
_profile.prepare_split_timer =
ADD_CHILD_TIMER_WITH_LEVEL(_scanner_profile, "PrepareSplitTime", table_profile, 1);
_profile.finalize_timer =
ADD_CHILD_TIMER_WITH_LEVEL(_scanner_profile, "FinalizeBlockTime", table_profile, 1);
_profile.create_reader_timer =
ADD_CHILD_TIMER_WITH_LEVEL(_scanner_profile, "CreateReaderTime", table_profile, 1);
_profile.pushdown_agg_timer =
ADD_CHILD_TIMER_WITH_LEVEL(_scanner_profile, "PushDownAggTime", table_profile, 1);
_profile.open_reader_timer =
ADD_CHILD_TIMER_WITH_LEVEL(_scanner_profile, "OpenReaderTime", table_profile, 1);
_profile.runtime_filter_partition_prune_timer = ADD_CHILD_TIMER_WITH_LEVEL(
_scanner_profile, "FileScannerRuntimeFilterPartitionPruningTime", table_profile, 1);
_profile.runtime_filter_partition_pruned_range_counter = ADD_CHILD_COUNTER_WITH_LEVEL(
_scanner_profile, "RuntimeFilterPartitionPrunedRangeNum", TUnit::UNIT,
table_profile, 1);
_profile.close_timer =
ADD_CHILD_TIMER_WITH_LEVEL(_scanner_profile, "CloseTime", table_profile, 1);
// Lifecycle timer names remain globally unique because RuntimeProfile's visual hierarchy
// does not namespace counters that share the same display parent.
_profile.file_reader_init_timer = ADD_CHILD_TIMER_WITH_LEVEL(
_scanner_profile, "FileReaderInitTime", file_reader_profile, 1);
_profile.file_reader_schema_timer = ADD_CHILD_TIMER_WITH_LEVEL(
_scanner_profile, "FileReaderGetSchemaTime", file_reader_profile, 1);
_profile.file_reader_mapper_timer = ADD_CHILD_TIMER_WITH_LEVEL(
_scanner_profile, "FileReaderCreateColumnMapperTime", file_reader_profile, 1);
_profile.file_reader_open_timer = ADD_CHILD_TIMER_WITH_LEVEL(
_scanner_profile, "FileReaderOpenTime", file_reader_profile, 1);
_profile.file_reader_get_block_timer = ADD_CHILD_TIMER_WITH_LEVEL(
_scanner_profile, "FileReaderGetBlockTime", file_reader_profile, 1);
_profile.file_reader_aggregate_timer = ADD_CHILD_TIMER_WITH_LEVEL(
_scanner_profile, "FileReaderAggregatePushDownTime", file_reader_profile, 1);
_profile.file_reader_close_timer = ADD_CHILD_TIMER_WITH_LEVEL(
_scanner_profile, "FileReaderCloseTime", file_reader_profile, 1);
}
// Establish lifecycle timers before consuming options or constructing filesystem properties;
// placing these scopes at the tail records only scope teardown and hides expensive init work.
SCOPED_TIMER(_profile.total_timer);
SCOPED_TIMER(_profile.init_timer);
_scan_params = options.scan_params;
_format = options.format;
_io_ctx = options.io_ctx;
_runtime_state = options.runtime_state;
_file_slot_descs = options.file_slot_descs;
_push_down_agg_type = options.push_down_agg_type;
_push_down_count_columns = options.push_down_count_columns;
_initial_condition_cache_digest = options.condition_cache_digest;
_condition_cache_digest = _initial_condition_cache_digest;
_projected_columns = std::move(options.projected_columns);
if (supports_iceberg_scan_semantics_v1(_scan_params)) {
for (auto& projected_column : _projected_columns) {
const auto* schema_field = find_external_root_field(_scan_params, projected_column);
if (schema_field != nullptr) {
attach_full_schema_identity(
&projected_column,
build_schema_identity_from_external_field(*schema_field));
}
}
}
_system_properties = create_system_properties(_scan_params);
_mapper_options.mode = TableColumnMappingMode::BY_NAME;
_conjuncts = std::move(options.conjuncts);
return Status::OK();
}
Status TableReader::_build_table_filters_from_conjuncts() {
_table_filters.clear();
_constant_pruning_safe_filter_count = 0;
bool in_safe_prefix = true;
for (const auto& conjunct : _conjuncts) {
DORIS_CHECK(conjunct != nullptr);
DORIS_CHECK(conjunct->root() != nullptr);
// `_table_filters` omits expressions without slot references, but such an expression still
// occupies a position in the row-level conjunct order. Record how many localized filters
// precede the first unsafe original conjunct so constant pruning cannot jump over a
// slotless non-deterministic/error-preserving barrier. Unsafe predicates remain solely on
// Scanner's original row-level path because localizing a clone would execute their state
// twice with independent state.
if (in_safe_prefix && !_is_safe_to_pre_execute(conjunct)) {
in_safe_prefix = false;
}
RETURN_IF_ERROR(
build_table_filters_from_conjunct(conjunct, _runtime_state, &_table_filters));
if (in_safe_prefix) {
_constant_pruning_safe_filter_count = _table_filters.size();
}
}
return Status::OK();
}
Status TableReader::_open_local_filter_exprs(const FileScanRequest& file_request) {
RowDescriptor row_desc;
for (const auto& conjunct : file_request.conjuncts) {
RETURN_IF_ERROR(conjunct->prepare(_runtime_state, row_desc));
RETURN_IF_ERROR(conjunct->open(_runtime_state));
}
for (const auto& delete_conjunct : file_request.delete_conjuncts) {
RETURN_IF_ERROR(delete_conjunct->prepare(_runtime_state, row_desc));
RETURN_IF_ERROR(delete_conjunct->open(_runtime_state));
}
return Status::OK();
}
bool TableReader::_should_enable_condition_cache(const FileScanRequest& file_request) const {
if (_condition_cache_digest == 0 || _push_down_agg_type == TPushAggOp::type::COUNT ||
_current_file_description == std::nullopt || _data_reader.reader == nullptr) {
return false;
}
// Condition cache is populated by file readers after evaluating file-local row-level
// conjuncts. Metadata pruning can skip row groups/pages, but it does not produce a per-row
// survivor bitmap that can safely populate the cache.
if (file_request.conjuncts.empty()) {
return false;
}
// Delete files/deletion vectors are table-format state. They may change independently of the
// data file path/mtime/size used by the external cache key, so caching their result can become
// stale. Keep delete filtering enabled, but do not read or write condition cache.
if (_delete_rows != nullptr || _deletion_vector != nullptr ||
!file_request.delete_conjuncts.empty()) {
return false;
}
// Only scanner-driven splits provide a digest rebuilt from the exact RF snapshot. Keep the
// conservative behavior for standalone TableReader callers: their initial digest may describe
// only static predicate P and must not store P AND RF under that key.
return _condition_cache_digest_covers_current_split ||
!contains_runtime_filter(file_request.conjuncts);
}
Status TableReader::_init_reader_condition_cache(const FileScanRequest& file_request) {
_condition_cache = nullptr;
_condition_cache_ctx = nullptr;
if (!_should_enable_condition_cache(file_request)) {
return Status::OK();
}
auto* cache = segment_v2::ConditionCache::instance();
if (cache == nullptr) {
return Status::OK();
}
const auto& file = *_current_file_description;
_condition_cache_key = segment_v2::ConditionCache::ExternalCacheKey(
file.path, file.mtime, file.file_size, _condition_cache_digest, file.range_start_offset,
file.range_size,
segment_v2::ConditionCache::ExternalCacheKey::BASE_GRANULE_AWARE_VERSION);
segment_v2::ConditionCacheHandle handle;
const bool condition_cache_hit = cache->lookup(_condition_cache_key, &handle);
if (condition_cache_hit) {
_condition_cache = handle.get_filter_result();
++_condition_cache_hit_count;
} else {
const int64_t total_rows = _data_reader.reader->get_total_rows();
if (total_rows <= 0) {
return Status::OK();
}
// Add one guard granule for split ranges that start in the middle of a granule. A guard
// false bit beyond the real range never overlaps real rows, but avoids boundary overflow
// when a reader marks the last partial granule.
const size_t num_granules = (total_rows + ConditionCacheContext::GRANULE_SIZE - 1) /
ConditionCacheContext::GRANULE_SIZE;
_condition_cache = std::make_shared<std::vector<bool>>(num_granules + 1, false);
}
if (_condition_cache != nullptr) {
_condition_cache_ctx = std::make_shared<ConditionCacheContext>();
_condition_cache_ctx->is_hit = condition_cache_hit;
_condition_cache_ctx->filter_result = _condition_cache;
_condition_cache_ctx->num_granules = _condition_cache->size();
if (condition_cache_hit) {
_condition_cache_ctx->base_granule = handle.get_base_granule();
}
_data_reader.reader->set_condition_cache_context(_condition_cache_ctx);
}
return Status::OK();
}
void TableReader::_finalize_reader_condition_cache() {
if (_condition_cache_ctx == nullptr || _condition_cache_ctx->is_hit) {
_condition_cache = nullptr;
_condition_cache_ctx = nullptr;
return;
}
// LIMIT or scanner cancellation may close a reader before all selected row ranges are visited.
// Unvisited granules remain false in a MISS bitmap, so inserting a partial bitmap would make a
// later HIT skip valid rows. Only publish cache entries after the physical reader reaches EOF.
if (!_current_reader_reached_eof) {
_condition_cache = nullptr;
_condition_cache_ctx = nullptr;
return;
}
DORIS_CHECK(_condition_cache_ctx->num_granules <= _condition_cache->size());
_condition_cache->resize(_condition_cache_ctx->num_granules);
segment_v2::ConditionCache::instance()->insert(
_condition_cache_key, std::move(_condition_cache), _condition_cache_ctx->base_granule);
_condition_cache = nullptr;
_condition_cache_ctx = nullptr;
}
Status TableReader::create_next_reader(bool* eos) {
SCOPED_TIMER(_profile.create_reader_timer);
DCHECK(_data_reader.reader == nullptr);
if (_current_task == nullptr) {
*eos = true;
return Status::OK();
}
RETURN_IF_ERROR(create_file_reader(&_data_reader.reader));
DORIS_CHECK(_data_reader.reader != nullptr);
if (_batch_size > 0) {
_data_reader.reader->set_batch_size(_batch_size);
}
Status st;
{
SCOPED_TIMER(_profile.file_reader_total_timer);
SCOPED_TIMER(_profile.file_reader_init_timer);
st = _data_reader.reader->init(_runtime_state);
}
if (!st.ok()) {
if (_io_ctx != nullptr && _io_ctx->should_stop && st.is<ErrorCode::END_OF_FILE>()) {
*eos = true;
_data_reader.reader.reset();
return Status::OK();
}
return st;
}
st = open_reader();
if (!st.ok()) {
if (_io_ctx != nullptr && _io_ctx->should_stop && st.is<ErrorCode::END_OF_FILE>()) {
*eos = true;
_data_reader.reader.reset();
return Status::OK();
}
return st;
}
if (_data_reader.reader == nullptr) {
*eos = _current_task == nullptr;
return Status::OK();
}
*eos = false;
return Status::OK();
}
Status TableReader::create_file_reader(std::unique_ptr<FileReader>* reader) {
DORIS_CHECK(reader != nullptr);
const bool enable_mapping_timestamp_tz = _scan_params != nullptr &&
_scan_params->__isset.enable_mapping_timestamp_tz &&
_scan_params->enable_mapping_timestamp_tz;
const bool enable_mapping_varbinary = _scan_params != nullptr &&
_scan_params->__isset.enable_mapping_varbinary &&
_scan_params->enable_mapping_varbinary;
if (_format == FileFormat::PARQUET) {
// V2 must honor the scan contract directly; otherwise Hive STRING columns backed by an
// unannotated BYTE_ARRAY are silently exposed as VARBINARY and predicate bytes no longer
// match the table type.
*reader = std::make_unique<format::parquet::ParquetReader>(
_system_properties, _current_task->data_file, _io_ctx, _scanner_profile,
_global_rowid_context, enable_mapping_timestamp_tz, enable_mapping_varbinary);
return Status::OK();
}
if (_format == FileFormat::ORC) {
*reader = std::make_unique<format::orc::OrcReader>(
_system_properties, _current_task->data_file, _io_ctx, _scanner_profile,
_global_rowid_context, enable_mapping_timestamp_tz);
return Status::OK();
}
if (_format == FileFormat::CSV) {
if (_file_slot_descs == nullptr) {
return Status::InvalidArgument("CSV reader requires file slot descriptors");
}
// CSV has no embedded schema. TableReader owns table-level mapping, while CsvReader needs
// only the physical file slots plus scan text parameters to build a file-local schema.
// Non-file columns such as partitions/defaults/virtual row ids are intentionally excluded
// from `_file_slot_descs` and are materialized during finalize_chunk().
*reader = std::make_unique<format::csv::CsvReader>(
_system_properties, _current_task->data_file, _io_ctx, _scanner_profile,
_scan_params, *_file_slot_descs, _current_range_compress_type,
_current_range_load_id);
return Status::OK();
}
if (_format == FileFormat::TEXT) {
if (_file_slot_descs == nullptr) {
return Status::InvalidArgument("Text reader requires file slot descriptors");
}
// Text files have no embedded schema. As with CSV, TableReader handles table-level mapping
// and only passes physical file slots to the v2 TextReader.
*reader = std::make_unique<format::text::TextReader>(
_system_properties, _current_task->data_file, _io_ctx, _scanner_profile,
_scan_params, *_file_slot_descs, _current_range_compress_type,
_current_range_load_id);
return Status::OK();
}
if (_format == FileFormat::JSON) {
if (_file_slot_descs == nullptr) {
return Status::InvalidArgument("JSON reader requires file slot descriptors");
}
*reader = std::make_unique<format::json::JsonReader>(
_system_properties, _current_task->data_file, _io_ctx, _scanner_profile,
_scan_params, _current_file_range_desc, *_file_slot_descs,
_current_range_compress_type, _current_range_load_id);
return Status::OK();
}
if (_format == FileFormat::NATIVE) {
*reader = std::make_unique<format::native::NativeReader>(
_system_properties, _current_task->data_file, _io_ctx, _scanner_profile);
return Status::OK();
}
return Status::NotSupported("TableReader does not support file format {}",
file_format_to_string(_format));
}
std::unique_ptr<io::FileDescription> create_file_description(const TFileRangeDesc& range) {
auto file_description = std::make_unique<io::FileDescription>();
file_description->path = range.path;
file_description->file_size = range.__isset.file_size ? range.file_size : -1;
file_description->mtime = range.__isset.modification_time ? range.modification_time : 0;
file_description->range_start_offset = range.__isset.start_offset ? range.start_offset : 0;
file_description->range_size = range.__isset.size ? range.size : -1;
if (range.__isset.fs_name) {
file_description->fs_name = range.fs_name;
}
if (range.__isset.file_cache_admission) {
file_description->file_cache_admission = range.file_cache_admission;
}
return file_description;
}
Status TableReader::prepare_split(const SplitReadOptions& options) {
SCOPED_TIMER(_profile.total_timer);
SCOPED_TIMER(_profile.prepare_split_timer);
_current_split_pruned = false;
_all_runtime_filters_applied_for_split = options.all_runtime_filters_applied;
_condition_cache_digest_covers_current_split = options.condition_cache_digest.has_value();
if (options.condition_cache_digest.has_value()) {
// The split snapshot may include RFs that arrived after TableReader::init(). Use the digest
// computed from that exact snapshot. Example: an initial P digest must not be used to store
// the bitmap for P AND late RF{7, 9}; the scanner supplies digest(P AND RF{7, 9}) here.
_condition_cache_digest = *options.condition_cache_digest;
} else {
// An explicit scanner digest is split-scoped. Restore the init-time digest when a later
// standalone split omits it instead of leaking the previous split's RF payload into its key.
_condition_cache_digest = _initial_condition_cache_digest;
}
if (options.conjuncts.has_value()) {
_conjuncts = *options.conjuncts;
}
// Update to current split format to handle ORC/PARQUET files in one table.
_format = options.current_split_format;
_partition_values = std::move(options.partition_values);
_current_task.reset();
_current_file_description.reset();
_current_file_range_desc = options.current_range;
_current_range_compress_type = options.current_range.__isset.compress_type
? options.current_range.compress_type
: TFileCompressType::UNKNOWN;
_current_range_load_id = options.current_range.__isset.load_id
? std::make_optional(options.current_range.load_id)
: std::nullopt;
_global_rowid_context = options.global_rowid_context;
_delete_rows = nullptr;
_deletion_vector = nullptr;
_aggregate_pushdown_tried = false;
_remaining_table_level_count = -1;
_remaining_file_level_count = -1;
_current_split_uses_metadata_count = false;
_current_reader_reached_eof = false;
RETURN_IF_ERROR(_evaluate_partition_prune_conjuncts(options.partition_prune_conjuncts,
&_current_split_pruned));
if (_current_split_pruned) {
COUNTER_UPDATE(_profile.runtime_filter_partition_pruned_range_counter, 1);
return Status::OK();
}
_current_task = std::make_unique<ScanTask>();
_current_task->data_file = create_file_description(options.current_range);
_current_file_description = *_current_task->data_file;
// A table-level row count is only equivalent to scanning the split when no row predicate is
// active and no predicate can arrive later. The metadata path can return several batches for
// one split; after its first synthetic batch there is no way to recover the real rows if a
// runtime filter arrives before the next scheduler turn.
// Table-level metadata only contains the number of rows; it cannot evaluate an expression or
// the NULL state of a COUNT argument. Require the new FE's explicit empty argument list, which
// means COUNT(*)/COUNT(1). A non-empty list means COUNT(col), while nullopt comes from an old FE
// whose COUNT semantics are unknown during a BE-first rolling upgrade.
if (_push_down_agg_type == TPushAggOp::type::COUNT && _push_down_count_columns.has_value() &&
_push_down_count_columns->empty() && options.all_runtime_filters_applied &&
_conjuncts.empty() && options.current_range.__isset.table_format_params &&
options.current_range.table_format_params.__isset.table_level_row_count) {
DORIS_CHECK(options.current_range.table_format_params.table_level_row_count >= -1);
_remaining_table_level_count =
options.current_range.table_format_params.table_level_row_count;
_current_split_uses_metadata_count = _is_table_level_count_active();
}
if (_is_table_level_count_active()) {
return Status::OK();
}
return _parse_delete_predicates(options);
}
Status TableReader::_evaluate_partition_prune_conjuncts(const VExprContextSPtrs& conjuncts,
bool* can_filter_all) {
DORIS_CHECK(can_filter_all != nullptr);
SCOPED_TIMER(_profile.runtime_filter_partition_prune_timer);
*can_filter_all = false;
if (conjuncts.empty() || _partition_values.empty()) {
return Status::OK();
}
VExprContextSPtrs partition_conjuncts;
for (const auto& conjunct : conjuncts) {
DORIS_CHECK(conjunct != nullptr);
DORIS_CHECK(conjunct->root() != nullptr);
// Keep only the safe prefix of the original conjunct order. If an unsafe conjunct is
// skipped, a later predicate could prune the split before the unsafe one reaches its
// normal row-level evaluation point.
if (!_is_safe_to_pre_execute(conjunct)) {
break;
}
std::set<GlobalIndex> global_indices;
collect_global_indices(conjunct->root(), &global_indices);
if (global_indices.empty()) {
continue;
}
const bool partition_only = std::ranges::all_of(global_indices, [&](GlobalIndex index) {
if (index.value() >= _projected_columns.size()) {
return false;
}
const auto& column = _projected_columns[index.value()];
return column.is_partition_key &&
find_partition_value(column, _partition_values) != nullptr;
});
if (partition_only) {
partition_conjuncts.push_back(conjunct);
}
}
if (partition_conjuncts.empty()) {
return Status::OK();
}
Block block;
RETURN_IF_ERROR(_build_partition_prune_block(&block));
RowDescriptor row_desc;
for (const auto& conjunct : partition_conjuncts) {
RETURN_IF_ERROR(conjunct->prepare(_runtime_state, row_desc));
RETURN_IF_ERROR(conjunct->open(_runtime_state));
}
IColumn::Filter result_filter(block.rows(), 1);
return VExprContext::execute_conjuncts(partition_conjuncts, nullptr, &block, &result_filter,
can_filter_all);
}
bool TableReader::_is_safe_to_pre_execute(const VExprContextSPtr& conjunct) {
DORIS_CHECK(conjunct != nullptr);
DORIS_CHECK(conjunct->root() != nullptr);
const auto root = conjunct->root();
const auto impl = root->get_impl();
const auto predicate = impl != nullptr ? impl : root;
// Split pruning evaluates a predicate once before any file rows are read. Reordering
// non-deterministic or error-preserving expressions can change their row-level semantics,
// even when every referenced slot is a partition column or maps to a constant entry.
return predicate->is_safe_to_execute_on_selected_rows();
}
Status TableReader::_build_partition_prune_block(Block* block) const {
DORIS_CHECK(block != nullptr);
DORIS_CHECK(!_projected_columns.empty());
block->clear();
for (const auto& column : _projected_columns) {
DORIS_CHECK(column.type != nullptr);
ColumnPtr value_column = column.type->create_column_const_with_default_value(1);
if (column.is_partition_key) {
const auto* partition_value = find_partition_value(column, _partition_values);
if (partition_value != nullptr) {
value_column = column.type->create_column_const(1, *partition_value);
}
}
block->insert({std::move(value_column), column.type, column.name});
}
return Status::OK();
}
Status TableReader::_parse_delete_predicates(const SplitReadOptions& options) {
DeleteFileDesc desc {.fs_name = options.current_range.fs_name};
bool has_delete_file = false;
RETURN_IF_ERROR(_parse_deletion_vector_file(options.current_range.table_format_params, &desc,
&has_delete_file));
if (has_delete_file) {
DORIS_CHECK(options.cache != nullptr);
Status create_status = Status::OK();
bool decoded_cache_hit = false;
_deletion_vector = options.cache->get<DeletionVector>(
desc.key,
[&]() -> DeletionVector* {
auto deletion_vector = std::make_unique<DeletionVector>();
DeletionVectorReader dv_reader(_runtime_state, _scanner_profile, *_scan_params,
desc, _io_ctx.get());
create_status = dv_reader.open();
if (!create_status.ok()) [[unlikely]] {
return nullptr;
}
size_t bytes_read = desc.size;
std::vector<char> buffer(bytes_read);
DBUG_EXECUTE_IF("TableReader.parse_deletion_vector.io_error", {
create_status =
Status::IOError("injected format v2 deletion vector read failure");
return nullptr;
});
DBUG_EXECUTE_IF("TableReader.parse_deletion_vector.should_stop", {
create_status = Status::EndOfFile("stop read.");
return nullptr;
});
create_status =
dv_reader.read_at(desc.start_offset, {buffer.data(), bytes_read});
const auto& file_cache_stats = dv_reader.file_cache_statistics();
COUNTER_UPDATE(_profile.dv_file_cache_hit_count,
file_cache_stats.num_local_io_total);
COUNTER_UPDATE(_profile.dv_file_cache_miss_count,
file_cache_stats.num_remote_io_total);
COUNTER_UPDATE(_profile.dv_file_cache_peer_read_count,
file_cache_stats.num_peer_io_total);
if (!create_status.ok()) [[unlikely]] {
return nullptr;
}
const char* buf = buffer.data();
SCOPED_TIMER(_profile.parse_delete_file_time);
create_status = parse_deletion_vector(buf, bytes_read, desc.format,
deletion_vector.get());
if (!create_status.ok()) [[unlikely]] {
return nullptr;
}
COUNTER_UPDATE(_profile.num_delete_rows, deletion_vector->cardinality());
return deletion_vector.release();
},
&decoded_cache_hit);
RETURN_IF_ERROR(create_status);
COUNTER_UPDATE(decoded_cache_hit ? _profile.decoded_dv_cache_hit_count
: _profile.decoded_dv_cache_miss_count,
1);
}
return Status::OK();
}
} // namespace doris::format