github-actions[bot] commented on code in PR #65559:
URL: https://github.com/apache/doris/pull/65559#discussion_r3594357331
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergUtils.java:
##########
@@ -1047,10 +1114,14 @@ public static List<Column> parseSchema(Schema schema,
boolean enableMappingVarbi
List<Types.NestedField> columns = schema.columns();
List<Column> resSchema =
Lists.newArrayListWithCapacity(columns.size());
for (Types.NestedField field : columns) {
- Column column = new Column(field.name().toLowerCase(Locale.ROOT),
+ String initialDefault = null;
Review Comment:
[P1] Preserve nested Iceberg initial defaults end to end
This loop assigns `initialDefault` only to top-level fields;
`updateIcebergColumnUniqueId()` recursively copies IDs but not defaults, so a
nested `INT` default never reaches the Thrift schema. Nested binary defaults
can arrive through the separate Base64 map, but the V2 access-path/child
mapping then drops them and materializes NULL/type defaults. Consequently old
files return and filter on the wrong value for newly added nested fields.
Please serialize defaults recursively here and preserve/parse them through the
BE child mapping, with old-file projection and predicate tests.
##########
be/src/core/data_type_serde/data_type_datev2_serde.cpp:
##########
@@ -126,6 +127,29 @@ Status
DataTypeDateV2SerDe::read_column_from_arrow(IColumn& column, const arrow:
return Status::OK();
}
+Status DataTypeDateV2SerDe::read_column_from_decoded_values(IColumn& column,
+ const
DecodedColumnView& view) const {
+ if (view.value_kind != DecodedValueKind::INT32) {
+ return decoded_column_view_handle_conversion_failure(
+ column, view, Status::NotSupported("DATEV2 decoded reader
expects INT32 source"));
+ }
+ if (view.values == nullptr &&
decoded_column_view_has_non_null_value(view)) {
+ return Status::Corruption("Decoded value buffer is null for {}",
column.get_name());
+ }
+ auto& data = assert_cast<ColumnDateV2&>(column).get_data();
+ const auto* values = reinterpret_cast<const int32_t*>(view.values);
+ for (int64_t row = 0; row < view.row_count; ++row) {
+ if (decoded_column_view_row_is_null(view, row)) {
+ data.push_back(DateV2Value<DateV2ValueType>());
+ continue;
+ }
+ DateV2Value<DateV2ValueType> date_v2;
+ date_v2.get_date_from_daynr(values[row] + date_threshold);
Review Comment:
[P1] Validate decoded Parquet DATE values before appending
`get_date_from_daynr()` returns false outside Doris's date range, but this
path ignores that result and unconditionally appends the default/invalid value.
The `int32_t` addition can also overflow for extreme, valid physical DATE
encodings. Consequently an out-of-range external date becomes a non-null bad
value and can change predicates/results instead of raising a data-quality error
or becoming NULL in nullable non-strict mode. Compute the day number in a wider
type and route range failures through the decoded conversion-failure policy.
##########
be/src/exec/operator/file_scan_operator.cpp:
##########
@@ -91,6 +104,28 @@ ScannerScheduler*
FileScanLocalState::scan_scheduler(RuntimeState* state) const
return state->get_query_ctx()->get_remote_scan_scheduler();
}
+#ifdef BE_TEST
+bool FileScanLocalState::TEST_should_use_file_scanner_v2(const TQueryOptions&
query_options,
+ bool is_load,
+ const
TFileScanRangeParams& scan_params) {
+ return _should_use_file_scanner_v2(query_options, is_load, scan_params);
+}
+#endif
+
+bool FileScanLocalState::_should_use_file_scanner_v2(const TQueryOptions&
query_options,
+ bool is_load,
+ const
TFileScanRangeParams& scan_params) {
+ const bool is_transactional_hive =
+ scan_params.__isset.table_format_params &&
+ scan_params.table_format_params.table_format_type ==
"transactional_hive";
+ // JNI reader selection is stored per split, but this scan-level selector
cannot inspect the
+ // split yet. Older FEs may omit both the scan-level Paimon marker and
split-level reader_type,
+ // so keep JNI scans on V1 until scanner selection can distinguish every
compatibility shape.
+ return query_options.__isset.enable_file_scanner_v2 &&
query_options.enable_file_scanner_v2 &&
Review Comment:
[P1] Keep unsupported query formats on the legacy scanner
With the new default-true flag, this scan-level predicate selects V2 for
every non-load format except WAL/JNI. Query AVRO and generic Arrow ranges
therefore reach `FileScannerV2::is_supported()`, which rejects AVRO and
non-remote-Doris Arrow on the first split, even though V1 has `AvroJNIReader`
and `ArrowStreamReader` implementations for them. Gate this selector with the
same complete capability matrix (or explicitly retain V1 for these formats) and
cover both in the selector test.
##########
be/src/format_v2/table_reader.cpp:
##########
@@ -0,0 +1,1024 @@
+// 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/iceberg_delete_file_reader_helper.h"
+#include "format/table/paimon_reader.h"
+#include "format_v2/column_mapper.h"
+#include "format_v2/deletion_vector_reader.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 "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
+ << ", 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();
+}
+
+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> {},
+ .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 = ¶ms->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;
+ }
+ for (const auto& field_ptr : schema->root_field.fields) {
+ const auto* field = get_field_ptr(field_ptr);
+ if (field == nullptr) {
+ continue;
+ }
+ if (external_field_matches_name(*field, 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);
+ column->identifier = context->schema_column->identifier;
+ column->name_mapping = context->schema_column->name_mapping;
+ 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) {
+ _scan_params = options.scan_params;
+ _format = options.format;
+ _io_ctx = options.io_ctx;
+ _runtime_state = options.runtime_state;
+ _scanner_profile = options.scanner_profile;
+ _file_slot_descs = options.file_slot_descs;
+ _push_down_agg_type = options.push_down_agg_type;
+ _condition_cache_digest = options.condition_cache_digest;
+ _projected_columns = std::move(options.projected_columns);
+ _system_properties = create_system_properties(_scan_params);
+ _mapper_options.mode = TableColumnMappingMode::BY_NAME;
+ _conjuncts = std::move(options.conjuncts);
+
+ if (_scanner_profile != nullptr) {
+ static const char* table_profile = "TableReader";
+ ADD_TIMER_WITH_LEVEL(_scanner_profile, 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_TIMER_WITH_LEVEL(
+ _scanner_profile,
"FileScannerRuntimeFilterPartitionPruningTime", 1);
+ _profile.runtime_filter_partition_pruned_range_counter =
ADD_COUNTER_WITH_LEVEL(
+ _scanner_profile, "RuntimeFilterPartitionPrunedRangeNum",
TUnit::UNIT, 1);
+ }
+ 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.
+ if (in_safe_prefix && !_is_safe_to_pre_execute(conjunct)) {
+ in_safe_prefix = false;
+ }
+ if (!in_safe_prefix) {
+ continue;
+ }
+ RETURN_IF_ERROR(
+ build_table_filters_from_conjunct(conjunct, _runtime_state,
&_table_filters));
+ _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;
+ }
+ // Runtime filters can arrive late and their payload is not guaranteed to
be represented by the
+ // scan-local digest. Without a read-only mode, a MISS could insert a
bitmap for P AND RF under
+ // the digest for only P. This mirrors the old FileScanner guard.
+ return !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 = _data_reader.reader->init(_runtime_state);
+ if (!st.ok()) {
+ if (_io_ctx != nullptr && _io_ctx->should_stop &&
st.is<ErrorCode::END_OF_FILE>()) {
Review Comment:
[P1] Treat reader EOF as an empty split
A zero-byte CSV/Text file deliberately returns `END_OF_FILE` from
`DelimitedTextReader::_open_file()`, and a valid header-only Native file
returns the same status during schema probing. This branch only converts EOF
when `should_stop` is set, so normal empty files escape through
`RETURN_IF_ERROR` and fail the entire FileScannerV2 query; the legacy scanner
skips such readers. Please close/reset the reader and finish this split on
non-cancellation EOF as well, with an empty-then-nonempty split regression.
##########
be/src/format_v2/table_reader.cpp:
##########
@@ -0,0 +1,1024 @@
+// 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/iceberg_delete_file_reader_helper.h"
+#include "format/table/paimon_reader.h"
+#include "format_v2/column_mapper.h"
+#include "format_v2/deletion_vector_reader.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 "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
+ << ", 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();
+}
+
+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> {},
+ .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 = ¶ms->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;
+ }
+ for (const auto& field_ptr : schema->root_field.fields) {
+ const auto* field = get_field_ptr(field_ptr);
+ if (field == nullptr) {
+ continue;
+ }
+ if (external_field_matches_name(*field, 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);
+ column->identifier = context->schema_column->identifier;
+ column->name_mapping = context->schema_column->name_mapping;
+ 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) {
+ _scan_params = options.scan_params;
+ _format = options.format;
+ _io_ctx = options.io_ctx;
+ _runtime_state = options.runtime_state;
+ _scanner_profile = options.scanner_profile;
+ _file_slot_descs = options.file_slot_descs;
+ _push_down_agg_type = options.push_down_agg_type;
+ _condition_cache_digest = options.condition_cache_digest;
+ _projected_columns = std::move(options.projected_columns);
+ _system_properties = create_system_properties(_scan_params);
+ _mapper_options.mode = TableColumnMappingMode::BY_NAME;
+ _conjuncts = std::move(options.conjuncts);
+
+ if (_scanner_profile != nullptr) {
+ static const char* table_profile = "TableReader";
+ ADD_TIMER_WITH_LEVEL(_scanner_profile, 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_TIMER_WITH_LEVEL(
+ _scanner_profile,
"FileScannerRuntimeFilterPartitionPruningTime", 1);
+ _profile.runtime_filter_partition_pruned_range_counter =
ADD_COUNTER_WITH_LEVEL(
+ _scanner_profile, "RuntimeFilterPartitionPrunedRangeNum",
TUnit::UNIT, 1);
+ }
+ 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.
+ if (in_safe_prefix && !_is_safe_to_pre_execute(conjunct)) {
+ in_safe_prefix = false;
+ }
+ if (!in_safe_prefix) {
+ continue;
+ }
+ RETURN_IF_ERROR(
+ build_table_filters_from_conjunct(conjunct, _runtime_state,
&_table_filters));
+ _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;
+ }
+ // Runtime filters can arrive late and their payload is not guaranteed to
be represented by the
+ // scan-local digest. Without a read-only mode, a MISS could insert a
bitmap for P AND RF under
+ // the digest for only P. This mirrors the old FileScanner guard.
+ return !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(
Review Comment:
[P1] Include storage identity in the external condition-cache key
This process-global key omits the filesystem/object-store configuration and
is enabled with `mtime == 0`. Iceberg/Hudi send zero mtime, while different
S3-compatible endpoints normalize to the same `s3://bucket/key`; two catalogs
can therefore have identical path/size/predicate keys but different bytes. A
false granule cached for one object is then skipped before residual filtering
on the other, silently losing qualifying rows. Include a stable
endpoint/filesystem plus object-version identity, or disable the cache when
they are unavailable as the new Parquet page cache does.
##########
be/src/core/data_type_serde/data_type_datetimev2_serde.cpp:
##########
@@ -44,6 +45,95 @@ namespace doris {
static const int64_t micro_to_nano_second = 1000;
#include "common/compile_check_begin.h"
+namespace {
+
+#pragma pack(1)
+struct DecodedInt96Timestamp {
+ int64_t nanos_of_day;
+ int32_t julian_day;
+
+ int64_t to_timestamp_micros() const {
+ static constexpr int32_t JULIAN_EPOCH_OFFSET_DAYS = 2440588;
+ static constexpr int64_t MICROS_IN_DAY = 86400000000;
+ static constexpr int64_t NANOS_PER_MICROSECOND = 1000;
+ return (julian_day - JULIAN_EPOCH_OFFSET_DAYS) * MICROS_IN_DAY +
+ nanos_of_day / NANOS_PER_MICROSECOND;
+ }
+};
+#pragma pack()
+static_assert(sizeof(DecodedInt96Timestamp) == 12);
+
+Status append_datetimev2_from_epoch_micros(ColumnDateTimeV2::Container& data,
+ int64_t timestamp_micros) {
+ static constexpr int64_t MICROS_PER_SECOND = 1000000;
+ static constexpr int64_t MICROS_PER_MINUTE = MICROS_PER_SECOND * 60;
+ static constexpr int64_t MICROS_PER_HOUR = MICROS_PER_MINUTE * 60;
+ static constexpr int64_t MICROS_PER_DAY = MICROS_PER_HOUR * 24;
+ static const int64_t EPOCH_DAYNR = calc_daynr(1970, 1, 1);
+
+ int64_t days_since_epoch = timestamp_micros / MICROS_PER_DAY;
+ int64_t micros_of_day = timestamp_micros % MICROS_PER_DAY;
+ if (micros_of_day < 0) {
+ micros_of_day += MICROS_PER_DAY;
+ --days_since_epoch;
+ }
+
+ const int64_t daynr = EPOCH_DAYNR + days_since_epoch;
+ if (daynr <= 0) {
+ return Status::DataQualityError(
+ "Decoded DATETIMEV2 timestamp is out of range: micros={},
daynr={}",
+ timestamp_micros, daynr);
+ }
+
+ DateV2Value<DateTimeV2ValueType> datetime_value;
+ if (!datetime_value.get_date_from_daynr(static_cast<uint64_t>(daynr))) {
+ return Status::DataQualityError(
+ "Decoded DATETIMEV2 timestamp is out of range: micros={},
daynr={}",
+ timestamp_micros, daynr);
+ }
+
+ const auto hour = static_cast<uint8_t>(micros_of_day / MICROS_PER_HOUR);
+ micros_of_day %= MICROS_PER_HOUR;
+ const auto minute = static_cast<uint8_t>(micros_of_day /
MICROS_PER_MINUTE);
+ micros_of_day %= MICROS_PER_MINUTE;
+ const auto second = static_cast<uint16_t>(micros_of_day /
MICROS_PER_SECOND);
+ const auto microsecond = static_cast<uint32_t>(micros_of_day %
MICROS_PER_SECOND);
+ datetime_value.unchecked_set_time(datetime_value.year(),
datetime_value.month(),
+ datetime_value.day(), hour, minute,
second, microsecond);
+ data.push_back(datetime_value);
+ return Status::OK();
+}
+
+void append_datetimev2_from_utc_epoch_micros(ColumnDateTimeV2::Container& data,
+ int64_t timestamp_micros,
+ const cctz::time_zone& timezone) {
+ static constexpr int64_t MICROS_PER_SECOND = 1000000;
+
+ int64_t epoch_seconds = timestamp_micros / MICROS_PER_SECOND;
+ int64_t micros_of_second = timestamp_micros % MICROS_PER_SECOND;
+ if (micros_of_second < 0) {
+ micros_of_second += MICROS_PER_SECOND;
+ --epoch_seconds;
+ }
+
+ DateV2Value<DateTimeV2ValueType> datetime_value;
+ datetime_value.from_unixtime(epoch_seconds, timezone);
+ datetime_value.set_microsecond(static_cast<uint32_t>(micros_of_second));
+ data.push_back(datetime_value);
+}
+
+int64_t decoded_timestamp_micros(const DecodedColumnView& view, int64_t value)
{
+ if (view.time_unit == DecodedTimeUnit::MILLIS) {
+ return value * 1000;
Review Comment:
[P1] Avoid overflow when converting Parquet milliseconds
`MILLIS` is an arbitrary signed `int64_t`, so multiplying by 1000 can
overflow before any Doris range check. For example, `INT64_MAX` commonly wraps
to `-1000` microseconds and is returned as a plausible 1969 value; the same
conversion also feeds TIMESTAMPTZ and statistics pruning. The legacy reader
avoids this intermediate by splitting the original value into seconds and
remainder. Use checked arithmetic or direct quotient/remainder conversion, then
route out-of-range values through the strict/nullable failure policy.
##########
be/src/format_v2/table/hive_reader.cpp:
##########
@@ -0,0 +1,246 @@
+// 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/hive_reader.h"
+
+#include <algorithm>
+#include <cctype>
+#include <string_view>
+
+#include "common/consts.h"
+#include "format_v2/column_mapper.h"
+#include "format_v2/file_reader.h"
+#include "runtime/runtime_state.h"
+
+namespace doris::format::hive {
+namespace {
+
+TFileFormatType::type format_type_from_context(const
format::ProjectedColumnBuildContext& context) {
+ DORIS_CHECK(context.scan_params != nullptr);
+ if (context.range != nullptr && context.range->__isset.format_type) {
+ return context.range->format_type;
+ }
+ return context.scan_params->format_type;
+}
+
+bool use_column_position_mapping(const format::ProjectedColumnBuildContext&
context) {
+ if (context.runtime_state == nullptr || context.scan_params == nullptr) {
+ return false;
+ }
+ switch (format_type_from_context(context)) {
+ case TFileFormatType::FORMAT_PARQUET:
+ return
!context.runtime_state->query_options().hive_parquet_use_column_names;
+ case TFileFormatType::FORMAT_ORC:
+ return
!context.runtime_state->query_options().hive_orc_use_column_names;
+ default:
+ return false;
+ }
+}
+
+bool is_file_column_position_slot(const TFileScanSlotInfo& slot_info,
+ const std::string& column_name) {
+ if (column_name.starts_with(BeConsts::GLOBAL_ROWID_COL) ||
+ column_name == BeConsts::ICEBERG_ROWID_COL) {
+ return false;
+ }
+ if (slot_info.__isset.is_file_slot) {
+ return slot_info.is_file_slot;
+ }
+ return true;
+}
+
+bool is_hive1_orc_column_name(std::string_view name) {
+ if (name.size() <= 4 || name.substr(0, 4) != "_col") {
+ return false;
+ }
+ return std::ranges::all_of(name.substr(4), [](unsigned char c) { return
std::isdigit(c); });
+}
+
+bool is_hive1_orc_file_schema(const std::vector<format::ColumnDefinition>&
file_schema) {
+ bool has_data_column = false;
+ for (const auto& field : file_schema) {
+ if (field.column_type != format::ColumnType::DATA_COLUMN) {
+ continue;
+ }
+ has_data_column = true;
+ if (!is_hive1_orc_column_name(field.name)) {
+ return false;
+ }
+ }
+ return has_data_column;
+}
+
+bool is_file_column_for_hive1_position_mapping(const format::ColumnDefinition&
column) {
+ if (column.column_type != format::ColumnType::DATA_COLUMN ||
column.is_partition_key) {
+ return false;
+ }
+ return !column.name.starts_with(BeConsts::GLOBAL_ROWID_COL) &&
+ column.name != BeConsts::ICEBERG_ROWID_COL;
+}
+
+void add_name_mapping(std::vector<std::string>* name_mapping, const
std::string& alias) {
+ DORIS_CHECK(name_mapping != nullptr);
+ if (std::ranges::find(*name_mapping, alias) == name_mapping->end()) {
+ name_mapping->push_back(alias);
+ }
+}
+
+} // namespace
+
+Status HiveReader::prepare_split(const format::SplitReadOptions& options) {
+ if (options.current_split_format != _format) {
+ return Status::InternalError(
+ "Hive scan expects all splits to use the same file format, "
+ "initialized_format={}, current_split_format={}",
+ static_cast<int>(_format),
static_cast<int>(options.current_split_format));
+ }
+ return format::TableReader::prepare_split(options);
+}
+
+format::TableColumnMappingMode HiveReader::mapping_mode() const {
+ // Hive-specific behavior: choose the column matching mode based on file
format and the
+ // matching session variable.
+ // - hive_orc_use_column_names / hive_parquet_use_column_names == true
+ // => BY_NAME (modern Hive default, match by column name)
+ // - those options == false
+ // => BY_INDEX (mainly for Hive1 ORC `_col0` / `_col1`, match by
top-level position;
+ // Parquet exposes the same switch for consistency)
+ // TableReader updates `_format` in prepare_split(), so this is evaluated
per split.
+ DORIS_CHECK(_runtime_state != nullptr);
+ const auto& query_options = _runtime_state->query_options();
+ bool use_column_names = true;
+ if (_format == format::FileFormat::ORC) {
+ use_column_names = query_options.hive_orc_use_column_names;
Review Comment:
[P1] Key condition-cache entries by the resolved Hive mapping
The condition digest is finalized before these options and the table schema
are resolved into file-local columns; Hive slot digests contain unique id `-1`
plus the logical name, not the chosen file ordinal. Thus both a mode toggle and
two positional Hive schemas that map logical `a` to different ordinals can
reuse the same path/version/predicate entry. A false bitmap from one binding
then skips rows before residual evaluation in the other. Include the resolved
table-to-file mapping/schema identity in the cache key or digest; hashing only
the two mode booleans is not sufficient.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
