github-actions[bot] commented on code in PR #63893:
URL: https://github.com/apache/doris/pull/63893#discussion_r3489040691
##########
regression-test/suites/external_table_p0/tvf/test_hdfs_parquet_group4.groovy:
##########
@@ -865,7 +865,7 @@ suite("test_hdfs_parquet_group4", "p0,external") {
"uri" = "${uri}",
"hadoop.username" = "${hdfsUserName}",
"format" = "parquet") limit 10; """
- exception "The column type of 'member0' is not supported"
+ exception "Parquet TIME with isAdjustedToUTC=true is not
supported"
Review Comment:
`git diff --check
a03283ae80337f7cf855a0d8e0ce0fef61fd7e6d..12ffec6d85e2030cdd8b21bedf3a2d72ab544cbc`
fails on this PR:
`regression-test/data/external_table_p0/tvf/test_hdfs_parquet_group4.out` has
trailing whitespace on lines 691, 1036, and 1395, and
`regression-test/data/external_table_p0/hive/write/test_hive_write_insert.out:217`
has a new blank line at EOF. Please strip the trailing tabs/remove the EOF
blank line so the expected outputs pass the standard whitespace check.
##########
regression-test/suites/external_table_p0/hive/test_hive_date_timezone.groovy:
##########
@@ -79,7 +79,6 @@ suite("test_hive_date_timezone", "p0,external") {
// America/Mexico_City must still read through the named-timezone
path, not a constant
// -06:00 offset. This fixture contains a 2022 DST timestamp that
makes the results differ.
assertEquals(parquetTimestampUtc.size(),
parquetTimestampMexicoCity.size())
Review Comment:
The comment still says `America/Mexico_City` must not behave like the
constant `-06:00` offset, but this now only checks that the named-zone result
has the same row count as UTC. The removed inequality against
`parquetTimestampFixedMexicoOffset` was the assertion that actually covered the
DST/named-timezone behavior, so a fixed-offset implementation would pass.
Please restore that assertion or replace it with an ordered expected-output
check proving the two result sets differ.
##########
be/src/format_v2/column_mapper.cpp:
##########
@@ -0,0 +1,2029 @@
+// 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/column_mapper.h"
+
+#include <algorithm>
+#include <cstddef>
+#include <memory>
+#include <sstream>
+#include <string_view>
+#include <utility>
+#include <vector>
+
+#include "common/consts.h"
+#include "common/exception.h"
+#include "common/status.h"
+#include "core/data_type/convert_field_to_type.h"
+#include "core/data_type/data_type_array.h"
+#include "core/data_type/data_type_map.h"
+#include "core/data_type/data_type_nullable.h"
+#include "core/data_type/data_type_string.h"
+#include "core/data_type/data_type_struct.h"
+#include "core/data_type/primitive_type.h"
+#include "exprs/runtime_filter_expr.h"
+#include "exprs/short_circuit_evaluation_expr.h"
+#include "exprs/vcase_expr.h"
+#include "exprs/vcast_expr.h"
+#include "exprs/vcondition_expr.h"
+#include "exprs/vectorized_fn_call.h"
+#include "exprs/vexpr_context.h"
+#include "exprs/vin_predicate.h"
+#include "exprs/vliteral.h"
+#include "format_v2/column_mapper_nested.h"
+#include "format_v2/expr/cast.h"
+#include "format_v2/file_reader.h"
+#include "format_v2/schema_projection.h"
+#include "format_v2/table_reader.h"
+#include "gen_cpp/Exprs_types.h"
+
+namespace doris::format {
+
+namespace {
+
+std::string mapping_mode_to_string(TableColumnMappingMode mode) {
+ switch (mode) {
+ case TableColumnMappingMode::BY_FIELD_ID:
+ return "BY_FIELD_ID";
+ case TableColumnMappingMode::BY_NAME:
+ return "BY_NAME";
+ case TableColumnMappingMode::BY_INDEX:
+ return "BY_INDEX";
+ }
+ return "UNKNOWN";
+}
+
+bool column_has_name(const ColumnDefinition& column, const std::string& name) {
+ if (to_lower(column.name) == to_lower(name)) {
+ return true;
+ }
+ if (column.has_identifier_name() && to_lower(column.get_identifier_name())
== to_lower(name)) {
+ return true;
+ }
+ return std::ranges::any_of(column.name_mapping, [&](const std::string&
alias) {
+ return to_lower(alias) == to_lower(name);
+ });
+}
+
+bool column_names_match(const ColumnDefinition& lhs, const ColumnDefinition&
rhs) {
+ if (column_has_name(rhs, lhs.name)) {
+ return true;
+ }
+ if (lhs.has_identifier_name() && column_has_name(rhs,
lhs.get_identifier_name())) {
+ return true;
+ }
+ return std::ranges::any_of(lhs.name_mapping, [&](const std::string& alias)
{
+ return column_has_name(rhs, alias);
+ });
+}
+
+class ColumnMatcher {
+public:
+ virtual ~ColumnMatcher() = default;
+ virtual const ColumnDefinition* find(
+ const ColumnDefinition& table_column,
+ const std::vector<ColumnDefinition>& file_schema) const = 0;
+};
+
+class FieldIdMatcher final : public ColumnMatcher {
+public:
+ const ColumnDefinition* find(const ColumnDefinition& table_column,
+ const std::vector<ColumnDefinition>&
file_schema) const override {
+ if (!table_column.has_identifier_field_id()) {
+ return nullptr;
+ }
+ const auto field_id = table_column.get_identifier_field_id();
+ const auto field_it = std::ranges::find_if(file_schema, [&](const
ColumnDefinition& field) {
+ return field.has_identifier_field_id() &&
field.get_identifier_field_id() == field_id;
+ });
+ return field_it == file_schema.end() ? nullptr : &*field_it;
+ }
+};
+
+class NameMatcher final : public ColumnMatcher {
+public:
+ const ColumnDefinition* find(const ColumnDefinition& table_column,
+ const std::vector<ColumnDefinition>&
file_schema) const override {
+ const auto field_it = std::ranges::find_if(file_schema, [&](const
ColumnDefinition& field) {
+ return column_names_match(table_column, field);
+ });
+ return field_it == file_schema.end() ? nullptr : &*field_it;
+ }
+};
+
+class PositionMatcher final : public ColumnMatcher {
+public:
+ const ColumnDefinition* find(const ColumnDefinition& table_column,
+ const std::vector<ColumnDefinition>&
file_schema) const override {
+ if (!table_column.has_identifier_field_id()) {
+ return nullptr;
+ }
+ const auto position = table_column.get_identifier_position();
+ if (position < 0 || static_cast<size_t>(position) >=
file_schema.size()) {
+ return nullptr;
+ }
+ return &file_schema[static_cast<size_t>(position)];
+ }
+};
+
+const ColumnMatcher& matcher_for_mode(TableColumnMappingMode mode) {
+ static const FieldIdMatcher field_id_matcher;
+ static const NameMatcher name_matcher;
+ static const PositionMatcher position_matcher;
+ switch (mode) {
+ case TableColumnMappingMode::BY_FIELD_ID:
+ return field_id_matcher;
+ case TableColumnMappingMode::BY_NAME:
+ return name_matcher;
+ case TableColumnMappingMode::BY_INDEX:
+ return position_matcher;
+ }
+ return field_id_matcher;
+}
+
+std::string virtual_column_type_to_string(TableVirtualColumnType type) {
+ switch (type) {
+ case TableVirtualColumnType::INVALID:
+ return "INVALID";
+ case TableVirtualColumnType::ROW_ID:
+ return "ROW_ID";
+ case TableVirtualColumnType::LAST_UPDATED_SEQUENCE_NUMBER:
+ return "LAST_UPDATED_SEQUENCE_NUMBER";
+ case TableVirtualColumnType::ICEBERG_ROWID:
+ return "ICEBERG_ROWID";
+ }
+ return "UNKNOWN";
+}
+
+std::string filter_conversion_type_to_string(FilterConversionType type) {
+ switch (type) {
+ case FilterConversionType::COPY_DIRECTLY:
+ return "COPY_DIRECTLY";
+ case FilterConversionType::CAST_FILTER:
+ return "CAST_FILTER";
+ case FilterConversionType::READER_EXPRESSION:
+ return "READER_EXPRESSION";
+ case FilterConversionType::FINALIZE_ONLY:
+ return "FINALIZE_ONLY";
+ case FilterConversionType::CONSTANT:
+ return "CONSTANT";
+ }
+ return "UNKNOWN";
+}
+
+std::string data_type_debug_string(const DataTypePtr& type) {
+ return type == nullptr ? "null" : type->get_name();
+}
+
+std::string field_debug_string(const Field& field) {
+ std::ostringstream out;
+ out << "Field{type=" << type_to_string(field.get_type()) << ", value=";
+ switch (field.get_type()) {
+ case TYPE_NULL:
+ out << "null";
+ break;
+ case TYPE_INT:
+ out << field.get<TYPE_INT>();
+ break;
+ case TYPE_BIGINT:
+ out << field.get<TYPE_BIGINT>();
+ break;
+ case TYPE_STRING:
+ out << field.get<TYPE_STRING>();
+ break;
+ default:
+ out << field.to_debug_string(0);
+ break;
+ }
+ out << "}";
+ return out.str();
+}
+
+template <typename T, typename Formatter>
+std::string join_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();
+}
+
+} // namespace
+
+const Field* find_partition_value(const ColumnDefinition& table_column,
+ const std::map<std::string, Field>&
partition_values) {
+ const auto find_by_name = [&](const std::string& name) -> const Field* {
+ const auto value_it = partition_values.find(name);
+ return value_it == partition_values.end() ? nullptr :
&value_it->second;
+ };
+ if (const auto* value = find_by_name(table_column.name); value != nullptr)
{
+ return value;
+ }
+ if (table_column.has_identifier_name()) {
+ if (const auto* value =
find_by_name(table_column.get_identifier_name());
+ value != nullptr) {
+ return value;
+ }
+ }
+ for (const auto& alias : table_column.name_mapping) {
+ if (const auto* value = find_by_name(alias); value != nullptr) {
+ return value;
+ }
+ }
+ return nullptr;
+}
+
+struct FileSlotRewriteInfo {
+ size_t block_position = 0;
+ DataTypePtr file_type;
+ DataTypePtr table_type;
+ std::string file_column_name;
+};
+
+struct RewriteContext {
+ RuntimeState* runtime_state = nullptr;
+ std::vector<VExprSPtr> created_exprs {};
+
+ void add_created_expr(VExprSPtr expr) {
created_exprs.push_back(std::move(expr)); }
+
+ Status prepare_created_exprs(VExprContext* context) const {
+ DORIS_CHECK(context != nullptr);
+ RowDescriptor row_desc;
+ for (const auto& expr : created_exprs) {
+ if (dynamic_cast<const Cast*>(expr.get()) != nullptr &&
runtime_state == nullptr) {
+ return Status::InvalidArgument(
+ "RuntimeState is required to prepare rewritten cast
expression {}",
+ expr->expr_name());
+ }
+ RETURN_IF_ERROR(expr->prepare(runtime_state, row_desc, context));
+ }
+ return Status::OK();
+ }
+};
+
+static VExprSPtr create_file_slot_ref(const VSlotRef& slot_ref,
+ const FileSlotRewriteInfo& rewrite_info,
+ RewriteContext* rewrite_context) {
+ auto ref =
+ VSlotRef::create_shared(slot_ref.slot_id(),
cast_set<int>(rewrite_info.block_position),
+ -1, rewrite_info.file_type,
rewrite_info.file_column_name);
+ rewrite_context->add_created_expr(ref);
+ return ref;
+}
+
+static bool is_cast_expr(const VExprSPtr& expr) {
+ return dynamic_cast<const Cast*>(expr.get()) != nullptr;
+}
+
+static bool is_binary_comparison_predicate(const VExprSPtr& expr) {
+ if (expr == nullptr || expr->get_num_children() != 2 ||
+ (expr->node_type() != TExprNodeType::BINARY_PRED &&
+ expr->node_type() != TExprNodeType::NULL_AWARE_BINARY_PRED)) {
+ return false;
+ }
+ switch (expr->op()) {
+ case TExprOpcode::EQ:
+ case TExprOpcode::EQ_FOR_NULL:
+ case TExprOpcode::NE:
+ case TExprOpcode::GE:
+ case TExprOpcode::GT:
+ case TExprOpcode::LE:
+ case TExprOpcode::LT:
+ return true;
+ default:
+ return false;
+ }
+}
+
+std::string TableColumnMapperOptions::debug_string() const {
+ std::ostringstream out;
+ out << "TableColumnMapperOptions{mode=" << mapping_mode_to_string(mode) <<
"}";
+ return out.str();
+}
+
+std::string ColumnDefinition::debug_string() const {
+ std::ostringstream out;
+ out << "ColumnDefinition{name=" << name << ", identifier=" <<
field_debug_string(identifier)
+ << ", name_mapping="
+ << join_debug_strings(name_mapping, [](const std::string& name) {
return name; })
+ << ", local_id=" << local_id << ", type=" <<
data_type_debug_string(type) << ", children="
+ << join_debug_strings(children,
+ [](const ColumnDefinition& child) { return
child.debug_string(); })
+ << ", has_default_expr=" << (default_expr != nullptr)
+ << ", is_partition_key=" << is_partition_key << "}";
+ return out.str();
+}
+
+std::string LocalColumnIndex::debug_string() const {
+ std::ostringstream out;
+ out << "LocalColumnIndex{index=" << index << ", project_all_children=" <<
project_all_children
+ << ", children="
+ << join_debug_strings(children,
+ [](const LocalColumnIndex& child) { return
child.debug_string(); })
+ << "}";
+ return out.str();
+}
+
+std::string ColumnMapping::debug_string() const {
+ std::ostringstream out;
+ out << "ColumnMapping{global_index=" << global_index
+ << ", table_column_name=" << table_column_name << ", file_local_id=";
+ if (file_local_id.has_value()) {
+ out << *file_local_id;
+ } else {
+ out << "null";
+ }
+ out << ", constant_index=";
+ if (constant_index.has_value()) {
+ out << *constant_index;
+ } else {
+ out << "null";
+ }
+ out << ", file_column_name=" << file_column_name
+ << ", original_file_type=" <<
data_type_debug_string(original_file_type)
+ << ", original_file_children="
+ << join_debug_strings(original_file_children,
+ [](const ColumnDefinition& child) { return
child.debug_string(); })
+ << ", file_type=" << data_type_debug_string(file_type)
+ << ", table_type=" << data_type_debug_string(table_type)
+ << ", has_projection=" << (projection != nullptr) << ",
child_mappings="
+ << join_debug_strings(child_mappings,
+ [](const ColumnMapping& child) { return
child.debug_string(); })
+ << ", is_trivial=" << is_trivial << ", is_constant=" <<
constant_index.has_value()
+ << ", filter_conversion=" <<
filter_conversion_type_to_string(filter_conversion)
+ << ", virtual_column_type=" <<
virtual_column_type_to_string(virtual_column_type)
+ << ", has_default_expr=" << (default_expr != nullptr) << "}";
+ return out.str();
+}
+
+std::string TableColumnMapper::debug_string() const {
+ std::ostringstream out;
+ out << "TableColumnMapper{options=" << _options.debug_string() << ",
mappings="
+ << join_debug_strings(_mappings,
+ [](const ColumnMapping& mapping) { return
mapping.debug_string(); })
+ << ", hidden_mappings="
+ << join_debug_strings(_hidden_mappings,
+ [](const ColumnMapping& mapping) { return
mapping.debug_string(); })
+ << ", constant_count=" << _constant_map.size() << "}";
+ return out.str();
+}
+
+static const FileSlotRewriteInfo* find_slot_rewrite_info(
+ const VExprSPtr& expr,
+ const std::map<GlobalIndex, FileSlotRewriteInfo>& global_to_file_slot,
+ const VSlotRef** slot_ref) {
+ if (expr == nullptr) {
+ return nullptr;
+ }
+ VExprSPtr slot_expr = expr;
+ const bool input_is_cast = is_cast_expr(expr) && expr->get_num_children()
== 1;
+ if (is_cast_expr(expr) && expr->get_num_children() == 1) {
+ slot_expr = expr->children()[0];
+ }
+ if (!slot_expr->is_slot_ref()) {
+ return nullptr;
+ }
+ const auto* candidate_slot_ref = assert_cast<const
VSlotRef*>(slot_expr.get());
+ const auto rewrite_it =
global_to_file_slot.find(slot_ref_global_index(*candidate_slot_ref));
+ if (rewrite_it == global_to_file_slot.end()) {
+ return nullptr;
+ }
+ if (input_is_cast &&
!expr->data_type()->equals(*rewrite_it->second.table_type)) {
+ return nullptr;
+ }
+ if (slot_ref != nullptr) {
+ *slot_ref = candidate_slot_ref;
+ }
+ return &rewrite_it->second;
+}
+
+static bool filter_conversion_has_local_source(FilterConversionType
conversion) {
+ switch (conversion) {
+ case FilterConversionType::COPY_DIRECTLY:
+ case FilterConversionType::CAST_FILTER:
+ case FilterConversionType::READER_EXPRESSION:
+ return true;
+ case FilterConversionType::FINALIZE_ONLY:
+ case FilterConversionType::CONSTANT:
+ return false;
+ }
+ return false;
+}
+
+static bool column_predicate_can_use_local_source(FilterConversionType
conversion) {
+ switch (conversion) {
+ case FilterConversionType::COPY_DIRECTLY:
+ return true;
+ case FilterConversionType::CAST_FILTER:
+ case FilterConversionType::READER_EXPRESSION:
+ case FilterConversionType::FINALIZE_ONLY:
+ case FilterConversionType::CONSTANT:
+ return false;
+ }
+ return false;
+}
+
+static bool table_filter_has_only_local_entries(
+ const TableFilter& table_filter, const std::map<GlobalIndex,
FilterEntry>& filter_entries) {
+ for (const auto global_index : table_filter.global_indices) {
+ const auto entry_it = filter_entries.find(global_index);
+ if (entry_it == filter_entries.end() || !entry_it->second.is_local()) {
+ return false;
+ }
+ }
+ return true;
+}
+
+static VExprSPtr unwrap_literal_for_file_cast(const VExprSPtr& expr,
+ const DataTypePtr& table_type) {
+ if (expr == nullptr) {
+ return nullptr;
+ }
+ if (expr->is_literal()) {
+ return expr;
+ }
+ if (is_cast_expr(expr) && expr->get_num_children() == 1 &&
expr->children()[0]->is_literal() &&
+ expr->children()[0]->data_type()->equals(*table_type)) {
+ return expr->children()[0];
+ }
+ return nullptr;
+}
+
+static Field literal_field_from_expr(const VExpr& literal_expr) {
+ DORIS_CHECK(literal_expr.is_literal());
+ const auto* literal = dynamic_cast<const VLiteral*>(&literal_expr);
+ DORIS_CHECK(literal != nullptr);
+ Field field;
+ literal->get_column_ptr()->get(0, field);
+ return field;
+}
+
+// Table filter localization clones an already-prepared table expr and then
rewrites it to file
+// slots. Only split-local literals and BE cast nodes need
table-reader-specific clone behavior;
+// plain slot refs and literals use their own VExpr::clone_node().
+static Status clone_table_expr_node(const VExpr& expr, VExprSPtr* cloned_expr)
{
+ DORIS_CHECK(cloned_expr != nullptr);
+ if (const auto* split_literal = dynamic_cast<const
SplitLocalFileLiteral*>(&expr)) {
+ *cloned_expr = std::make_shared<SplitLocalFileLiteral>(
+ split_literal->data_type(), literal_field_from_expr(expr),
+ split_literal->original_type(),
split_literal->original_field());
+ } else if (const auto* vcast_expr = dynamic_cast<const VCastExpr*>(&expr);
+ vcast_expr != nullptr && vcast_expr->node_type() ==
TExprNodeType::CAST_EXPR) {
+ *cloned_expr = Cast::create_shared(vcast_expr->data_type());
+ }
+ return Status::OK();
+}
+
+Status clone_table_expr_tree(const VExprSPtr& expr, VExprSPtr* cloned_expr) {
+ DORIS_CHECK(cloned_expr != nullptr);
+ if (expr == nullptr) {
+ *cloned_expr = nullptr;
+ return Status::OK();
+ }
+ return expr->deep_clone(cloned_expr, clone_table_expr_node);
+}
+
+static VExprSPtr original_table_literal(const VExprSPtr& literal_expr,
+ RewriteContext* rewrite_context =
nullptr) {
+ DORIS_CHECK(literal_expr != nullptr);
+ DORIS_CHECK(literal_expr->is_literal());
+ const auto* rewritten_literal = dynamic_cast<const
SplitLocalFileLiteral*>(literal_expr.get());
+ if (rewritten_literal == nullptr) {
+ return literal_expr;
+ }
+ auto literal = VLiteral::create_shared(rewritten_literal->original_type(),
+
rewritten_literal->original_field());
+ if (rewrite_context != nullptr) {
+ rewrite_context->add_created_expr(literal);
+ }
+ return literal;
+}
+
+static ColumnDefinition hidden_column_from_slot_ref(const VSlotRef& slot_ref) {
+ ColumnDefinition column;
+ column.name = slot_ref.column_name();
+ column.identifier = Field::create_field<TYPE_STRING>(column.name);
+ column.type = slot_ref.data_type();
+ return column;
+}
+
+static void collect_top_level_slot_columns(const VExprSPtr& expr,
+ std::map<GlobalIndex,
ColumnDefinition>* columns) {
+ DORIS_CHECK(columns != nullptr);
+ if (expr == nullptr) {
+ return;
+ }
+ if (expr->is_slot_ref()) {
+ const auto* slot_ref = assert_cast<const VSlotRef*>(expr.get());
+ columns->try_emplace(slot_ref_global_index(*slot_ref),
+ hidden_column_from_slot_ref(*slot_ref));
+ return;
+ }
+ for (const auto& child : expr->children()) {
+ collect_top_level_slot_columns(child, columns);
+ }
+}
+
+static VExprSPtr rewrite_literal_to_file_type(const VExprSPtr& literal_expr,
+ const FileSlotRewriteInfo&
rewrite_info,
+ RewriteContext* rewrite_context)
{
+ DORIS_CHECK(literal_expr != nullptr);
+ DORIS_CHECK(literal_expr->is_literal());
+ const auto original_literal = original_table_literal(literal_expr,
rewrite_context);
+ const Field original_field = literal_field(original_literal);
+ if (rewrite_info.file_type->equals(*original_literal->data_type())) {
+ return original_literal;
+ }
+ Field file_field;
+ try {
+ convert_field_to_type(original_field, *rewrite_info.file_type,
&file_field,
+ original_literal->data_type().get());
+ } catch (const Exception&) {
+ return nullptr;
+ }
+ if (file_field.is_null()) {
+ return nullptr;
+ }
+ if (file_field.get_type() !=
remove_nullable(rewrite_info.file_type)->get_primitive_type()) {
+ return nullptr;
+ }
+ auto literal = std::make_shared<SplitLocalFileLiteral>(
+ rewrite_info.file_type, file_field, original_literal->data_type(),
original_field);
+ rewrite_context->add_created_expr(literal);
+ return literal;
+}
+
+static bool rewrite_binary_slot_literal_predicate(
+ const VExprSPtr& expr,
+ const std::map<GlobalIndex, FileSlotRewriteInfo>& global_to_file_slot,
+ RewriteContext* rewrite_context) {
+ if (!is_binary_comparison_predicate(expr)) {
+ return false;
+ }
+ auto children = expr->children();
+ const VSlotRef* slot_ref = nullptr;
+ const FileSlotRewriteInfo* rewrite_info =
+ find_slot_rewrite_info(children[0], global_to_file_slot,
&slot_ref);
+ int slot_child_idx = 0;
+ int literal_child_idx = 1;
+ if (rewrite_info == nullptr) {
+ rewrite_info = find_slot_rewrite_info(children[1],
global_to_file_slot, &slot_ref);
+ slot_child_idx = 1;
+ literal_child_idx = 0;
+ }
+ if (rewrite_info == nullptr || slot_ref == nullptr) {
+ return false;
+ }
+ auto literal_expr =
+ unwrap_literal_for_file_cast(children[literal_child_idx],
rewrite_info->table_type);
+ if (literal_expr == nullptr) {
+ return false;
+ }
+
+ auto rewritten_literal =
+ rewrite_literal_to_file_type(literal_expr, *rewrite_info,
rewrite_context);
+ if (rewritten_literal == nullptr) {
+ children[literal_child_idx] = original_table_literal(literal_expr,
rewrite_context);
+ expr->set_children(std::move(children));
+ return false;
+ }
+
+ children[slot_child_idx] = create_file_slot_ref(*slot_ref, *rewrite_info,
rewrite_context);
+ children[literal_child_idx] = std::move(rewritten_literal);
+ expr->set_children(std::move(children));
+ return true;
+}
+
+static bool rewrite_in_slot_literal_predicate(
+ const VExprSPtr& expr,
+ const std::map<GlobalIndex, FileSlotRewriteInfo>& global_to_file_slot,
+ RewriteContext* rewrite_context) {
+ if (expr->node_type() != TExprNodeType::IN_PRED ||
expr->get_num_children() < 2) {
+ return false;
+ }
+ auto children = expr->children();
+ const VSlotRef* slot_ref = nullptr;
+ const FileSlotRewriteInfo* rewrite_info =
+ find_slot_rewrite_info(children[0], global_to_file_slot,
&slot_ref);
+ if (rewrite_info == nullptr || slot_ref == nullptr) {
+ return false;
+ }
+
+ VExprSPtrs rewritten_literals;
+ rewritten_literals.reserve(children.size() - 1);
+ for (size_t child_idx = 1; child_idx < children.size(); ++child_idx) {
+ auto literal_expr =
+ unwrap_literal_for_file_cast(children[child_idx],
rewrite_info->table_type);
+ if (literal_expr == nullptr) {
+ return false;
+ }
+ auto rewritten_literal =
+ rewrite_literal_to_file_type(literal_expr, *rewrite_info,
rewrite_context);
+ if (rewritten_literal == nullptr) {
+ for (size_t restore_idx = 1; restore_idx < children.size();
++restore_idx) {
+ auto restore_literal =
unwrap_literal_for_file_cast(children[restore_idx],
+
rewrite_info->table_type);
+ if (restore_literal != nullptr) {
+ children[restore_idx] =
+ original_table_literal(restore_literal,
rewrite_context);
+ }
+ }
+ expr->set_children(std::move(children));
+ return false;
+ }
+ rewritten_literals.push_back(std::move(rewritten_literal));
+ }
+
+ children[0] = create_file_slot_ref(*slot_ref, *rewrite_info,
rewrite_context);
+ for (size_t literal_idx = 0; literal_idx < rewritten_literals.size();
++literal_idx) {
+ children[literal_idx + 1] = std::move(rewritten_literals[literal_idx]);
+ }
+ expr->set_children(std::move(children));
+ return true;
+}
+
+static VExprSPtr create_file_struct_child_name_literal(const std::string&
file_child_name,
+ RewriteContext*
rewrite_context) {
+ auto literal = VLiteral::create_shared(std::make_shared<DataTypeString>(),
+
Field::create_field<TYPE_STRING>(file_child_name));
+ rewrite_context->add_created_expr(literal);
+ return literal;
+}
+
+static bool needs_complex_file_slot_cast(const DataTypePtr& file_type,
+ const DataTypePtr& table_type) {
+ if (file_type == nullptr || table_type == nullptr ||
file_type->equals(*table_type)) {
+ return false;
+ }
+ const auto file_nested_type = remove_nullable(file_type);
+ const auto table_nested_type = remove_nullable(table_type);
+ if (file_nested_type->equals(*table_nested_type)) {
+ return false;
+ }
+ return is_complex_type(file_nested_type->get_primitive_type()) ||
+ is_complex_type(table_nested_type->get_primitive_type());
+}
+
+static bool collect_struct_element_chain(const VExprSPtr& expr,
std::vector<VExprSPtr>* chain) {
+ DORIS_CHECK(chain != nullptr);
+ if (!is_struct_element_expr(expr)) {
+ return false;
+ }
+ const auto& parent = expr->children()[0];
+ if (is_struct_element_expr(parent)) {
+ if (!collect_struct_element_chain(parent, chain)) {
+ return false;
+ }
+ } else if (!parent->is_slot_ref()) {
+ // Only support file-local rewrite for struct child chains rooted
directly at a top-level
+ // slot, for example `element_at(s, 'a')` or `element_at(element_at(s,
'a'), 'b')`.
+ //
+ // Do not localize computed complex parents such as
+ // `element_at(element_at(map_values(m), 1), 'full_name')`. The
intermediate map/array
+ // result has already been reshaped by scan projection and may have a
different child order
+ // from the table expression. Partially rewriting that expression
against the file block can
+ // silently evaluate the wrong struct child and filter out valid rows.
Those predicates must
+ // remain as table-level conjuncts and be evaluated after TableReader
materialization.
+ return false;
+ }
+ chain->push_back(expr);
+ return true;
+}
+
+static bool rewrite_struct_element_path_to_file_expr(
+ const VExprSPtr& expr, const std::vector<ColumnMapping>& mappings,
+ const std::map<GlobalIndex, FileSlotRewriteInfo>& global_to_file_slot,
+ RewriteContext* rewrite_context) {
+ ResolvedNestedStructPath resolved;
+ if (!resolve_nested_struct_expr_for_file(expr, mappings, &resolved)) {
+ return false;
+ }
+
+ std::vector<VExprSPtr> struct_element_chain;
+ if (!collect_struct_element_chain(expr, &struct_element_chain) ||
+ struct_element_chain.size() != resolved.file_child_names.size() ||
+ struct_element_chain.size() != resolved.file_child_types.size()) {
+ return false;
+ }
+
+ auto root_children = struct_element_chain.front()->children();
+ if (!root_children[0]->is_slot_ref()) {
+ return false;
+ }
+ const auto* slot_ref = assert_cast<const
VSlotRef*>(root_children[0].get());
+ const auto rewrite_it =
global_to_file_slot.find(slot_ref_global_index(*slot_ref));
+ if (rewrite_it == global_to_file_slot.end()) {
+ return false;
+ }
+
+ // File-local conjuncts are prepared against the file-reader Block, so
both the root slot and
+ // every struct selector must be expressed in file schema terms. For a
renamed Iceberg field,
+ // keeping the table selector would prepare
`element_at(file_struct<rename>, 'renamed')` and
+ // fail before any rows are read. Rewrite the whole chain while
ColumnMapping still preserves
+ // the table-to-file relationship. Example:
+ // table filter: element_at(element_at(s, 'renamed_parent'),
'renamed_leaf')
+ // old file: s<parent<leaf>>
+ // file filter: element_at(element_at(s, 'parent'), 'leaf')
+ root_children[0] = create_file_slot_ref(*slot_ref, rewrite_it->second,
rewrite_context);
+ struct_element_chain.front()->set_children(std::move(root_children));
+ for (size_t idx = 0; idx < struct_element_chain.size(); ++idx) {
+ auto children = struct_element_chain[idx]->children();
+ children[1] =
create_file_struct_child_name_literal(resolved.file_child_names[idx],
+ rewrite_context);
+ struct_element_chain[idx]->set_children(std::move(children));
+ // The selector name and the expression return type must be moved to
file schema together.
+ // Example:
+ // table filter: element_at(element_at(s, 'new_a'), 'new_aa') = 50
+ // old file: s.new_a STRUCT<aa, bb>
+ // file filter: element_at(element_at(s, 'new_a'), 'aa') = 50
+ //
+ // If the inner element_at keeps the table return type STRUCT<new_aa,
bb>, preparing the
+ // outer element_at(..., 'aa') fails before scanning because `aa` is
not a table field.
+ struct_element_chain[idx]->data_type() =
resolved.file_child_types[idx];
+ }
+ return true;
+}
+
+static VExprSPtr rewrite_table_expr_to_file_expr(
+ const VExprSPtr& expr,
+ const std::map<GlobalIndex, FileSlotRewriteInfo>& global_to_file_slot,
+ const std::vector<ColumnMapping>& filter_mappings, RewriteContext*
rewrite_context,
+ bool* can_localize) {
+ if (expr == nullptr) {
+ return nullptr;
+ }
+ DORIS_CHECK(rewrite_context != nullptr);
+ DORIS_CHECK(can_localize != nullptr);
+ if (auto* runtime_filter = dynamic_cast<RuntimeFilterExpr*>(expr.get());
+ runtime_filter != nullptr) {
+ auto impl = runtime_filter->get_impl();
+ if (impl == nullptr) {
+ *can_localize = false;
+ return expr;
+ }
+ auto localized_impl = rewrite_table_expr_to_file_expr(
+ impl, global_to_file_slot, filter_mappings, rewrite_context,
can_localize);
+ if (!*can_localize) {
+ return expr;
+ }
+ runtime_filter->set_impl(std::move(localized_impl));
+ return expr;
+ }
+ if (rewrite_binary_slot_literal_predicate(expr, global_to_file_slot,
rewrite_context)) {
+ return expr;
+ }
+ if (rewrite_in_slot_literal_predicate(expr, global_to_file_slot,
rewrite_context)) {
+ return expr;
+ }
+ if (is_struct_element_expr(expr)) {
+ if (!rewrite_struct_element_path_to_file_expr(expr, filter_mappings,
global_to_file_slot,
+ rewrite_context)) {
+ // The scanner still evaluates the original table-level conjunct
after TableReader
+ // finalizes the output block. Skipping an unlocalizable file
conjunct is therefore
+ // safer than preparing a partially rewritten expression against
the wrong struct
+ // layout. In particular, do not generate file-local conjuncts for
computed complex
+ // parents such as `element_at(element_at(map_values(m), 1),
'field')`; only direct
+ // slot-rooted struct chains are supported here.
+ *can_localize = false;
+ }
+ return expr;
+ }
+ if (expr->is_slot_ref()) {
+ const auto* slot_ref = assert_cast<const VSlotRef*>(expr.get());
+ const auto rewrite_it =
global_to_file_slot.find(slot_ref_global_index(*slot_ref));
+ if (rewrite_it != global_to_file_slot.end()) {
+ const auto& rewrite_info = rewrite_it->second;
+ auto file_slot = create_file_slot_ref(*slot_ref, rewrite_info,
rewrite_context);
+ if (rewrite_info.file_type->equals(*rewrite_info.table_type)) {
+ return file_slot;
+ }
+ if (needs_complex_file_slot_cast(rewrite_info.file_type,
rewrite_info.table_type)) {
+ // Generic file-local expressions cannot safely cast an
evolved complex file slot
+ // back to the table type. Example:
+ //
+ // table filter: ARRAY_CONTAINS(MAP_KEYS(m), 'person5')
+ // old file: m MAP<STRING, STRUCT<name, age>>
+ // table: m MAP<STRING, STRUCT<age, full_name,
gender>>
+ //
+ // Although MAP_KEYS only reads the key column, wrapping the
file slot as
+ // `CAST(file_m AS table_m)` forces the value struct cast
first and fails because
+ // the old and new value structs have different fields. Keep
such filters at the
+ // table level, where TableReader materializes the evolved
complex value before
+ // Scanner evaluates the original conjunct. Direct slot-rooted
struct child paths
+ // are handled by rewrite_struct_element_path_to_file_expr()
above.
+ *can_localize = false;
+ return expr;
+ }
+ auto cast_expr = Cast::create_shared(rewrite_info.table_type);
+ cast_expr->add_child(std::move(file_slot));
+ rewrite_context->add_created_expr(cast_expr);
+ return cast_expr;
+ }
+ return expr;
+ }
+ // The input is a split-local cloned tree. A previous split-local clone
may already have
+ // inserted Cast(slot). Keep that rewrite idempotent: rewrite the cast
child from table slot to
+ // the current split's file slot, and drop the cast when the current split
no longer needs it.
+ if (is_cast_expr(expr) && expr->get_num_children() == 1) {
+ const auto& child = expr->children()[0];
+ if (child->is_slot_ref()) {
+ const auto* slot_ref = assert_cast<const VSlotRef*>(child.get());
+ const auto rewrite_it =
global_to_file_slot.find(slot_ref_global_index(*slot_ref));
+ if (rewrite_it != global_to_file_slot.end() &&
+ expr->data_type()->equals(*rewrite_it->second.table_type)) {
+ auto rewritten_child =
+ create_file_slot_ref(*slot_ref, rewrite_it->second,
rewrite_context);
+ if
(rewrite_it->second.file_type->equals(*rewrite_it->second.table_type)) {
+ return rewritten_child;
+ }
+ if (needs_complex_file_slot_cast(rewrite_it->second.file_type,
+
rewrite_it->second.table_type)) {
+ *can_localize = false;
+ return expr;
+ }
+ expr->set_children({std::move(rewritten_child)});
+ return expr;
+ }
+ }
+ }
+
+ VExprSPtrs rewritten_children;
+ rewritten_children.reserve(expr->children().size());
+ for (const auto& child : expr->children()) {
+ rewritten_children.push_back(rewrite_table_expr_to_file_expr(
+ child, global_to_file_slot, filter_mappings, rewrite_context,
can_localize));
+ }
+ expr->set_children(std::move(rewritten_children));
+ return expr;
+}
+
+static constexpr const char* ROW_LINEAGE_ROW_ID = "_row_id";
+static constexpr const char* ROW_LINEAGE_LAST_UPDATED_SEQ_NUMBER =
"_last_updated_sequence_number";
+static constexpr int32_t ROW_LINEAGE_ROW_ID_FIELD_ID = 2147483540;
+static constexpr int32_t ROW_LINEAGE_LAST_UPDATED_SEQ_NUMBER_FIELD_ID =
2147483539;
+
+static TableVirtualColumnType row_lineage_virtual_column_type(const
std::string& column_name) {
+ if (column_name == ROW_LINEAGE_ROW_ID) {
+ return TableVirtualColumnType::ROW_ID;
+ }
+ if (column_name == ROW_LINEAGE_LAST_UPDATED_SEQ_NUMBER) {
+ return TableVirtualColumnType::LAST_UPDATED_SEQUENCE_NUMBER;
+ }
+ return TableVirtualColumnType::INVALID;
+}
+
+static TableVirtualColumnType row_lineage_virtual_column_type_by_field_id(
+ const ColumnDefinition& column) {
+ if (!column.has_identifier_field_id()) {
+ return TableVirtualColumnType::INVALID;
+ }
+ switch (column.get_identifier_field_id()) {
+ case ROW_LINEAGE_ROW_ID_FIELD_ID:
+ return TableVirtualColumnType::ROW_ID;
+ case ROW_LINEAGE_LAST_UPDATED_SEQ_NUMBER_FIELD_ID:
+ return TableVirtualColumnType::LAST_UPDATED_SEQUENCE_NUMBER;
+ default:
+ return TableVirtualColumnType::INVALID;
+ }
+}
+
+static TableVirtualColumnType row_lineage_virtual_column_type(const
ColumnDefinition& column,
+
TableColumnMappingMode mode) {
+ switch (mode) {
+ case TableColumnMappingMode::BY_FIELD_ID:
+ return row_lineage_virtual_column_type_by_field_id(column);
+ case TableColumnMappingMode::BY_NAME:
+ case TableColumnMappingMode::BY_INDEX:
+ return row_lineage_virtual_column_type(column.name);
+ }
+ return TableVirtualColumnType::INVALID;
+}
+
+// Returns true when the current file type is not the exact nested type the
scan should expose.
+// This is about building the projected file-side type/projection, not about
whether TableReader
+// later needs to rematerialize the complex value back to table layout.
+static bool needs_projected_file_type_rebuild(const ColumnMapping& mapping) {
+ if (!is_complex_type(mapping.file_type->get_primitive_type())) {
+ return false;
+ }
+ if (mapping.child_mappings.empty()) {
+ return false;
+ }
+ DORIS_CHECK(mapping.file_type != nullptr);
+ DORIS_CHECK(mapping.table_type != nullptr);
+ if (remove_nullable(mapping.file_type)->get_primitive_type() !=
+ remove_nullable(mapping.table_type)->get_primitive_type()) {
+ return true;
+ }
+ if (!mapping.table_type->equals(*mapping.file_type)) {
+ return true;
+ }
+ for (const auto& child_mapping : mapping.child_mappings) {
+ // Rename-only child mappings do not change the file-side projected
shape. If field-id
+ // matching maps table child `renamed_b` to file child `b`, the file
reader can still expose
+ // the original file type as long as child count/order/types are
unchanged.
+ if (!child_mapping.file_local_id.has_value() ||
+ needs_projected_file_type_rebuild(child_mapping)) {
+ return true;
+ }
+ }
+ return false;
+}
+
+static std::optional<size_t> file_child_ordinal_in_scan_type(const
ColumnMapping& mapping,
+ const
ColumnMapping& child_mapping) {
+ if (!child_mapping.file_local_id.has_value()) {
+ return std::nullopt;
+ }
+ const auto& file_children = !mapping.projected_file_children.empty()
+ ? mapping.projected_file_children
+ : mapping.original_file_children;
+ const auto child_it = std::ranges::find_if(file_children, [&](const
ColumnDefinition& child) {
+ return child.file_local_id() == *child_mapping.file_local_id;
+ });
+ if (child_it == file_children.end()) {
+ return std::nullopt;
+ }
+ return static_cast<size_t>(std::distance(file_children.begin(), child_it));
+}
+
+static bool needs_complex_rematerialize(const ColumnMapping& mapping) {
+ if (mapping.child_mappings.empty()) {
+ return false;
+ }
+ if (mapping.table_type == nullptr || mapping.file_type == nullptr ||
+ !mapping.table_type->equals(*mapping.file_type)) {
+ return true;
+ }
+ for (size_t table_child_idx = 0; table_child_idx <
mapping.child_mappings.size();
+ ++table_child_idx) {
+ const auto& child_mapping = mapping.child_mappings[table_child_idx];
+ const auto file_child_idx = file_child_ordinal_in_scan_type(mapping,
child_mapping);
+ if (!file_child_idx.has_value() || *file_child_idx != table_child_idx
||
+ needs_complex_rematerialize(child_mapping) ||
+ (child_mapping.table_type != nullptr && child_mapping.file_type !=
nullptr &&
+ !child_mapping.table_type->equals(*child_mapping.file_type))) {
+ return true;
+ }
+ }
+ return false;
+}
+
+static bool mapping_can_use_file_column_directly(const ColumnMapping& mapping)
{
+ if (mapping.table_type == nullptr || mapping.file_type == nullptr) {
+ return false;
+ }
+ const auto table_type = remove_nullable(mapping.table_type);
+ const auto file_type = remove_nullable(mapping.file_type);
+ const bool same_timestamptz_with_different_scale =
+ table_type->get_primitive_type() == TYPE_TIMESTAMPTZ &&
+ file_type->get_primitive_type() == TYPE_TIMESTAMPTZ;
+ if (!mapping.table_type->equals(*mapping.file_type) &&
!same_timestamptz_with_different_scale) {
+ return false;
+ }
+ return !needs_complex_rematerialize(mapping);
+}
+
+static const ColumnDefinition* find_file_child_for_mapping(const
ColumnDefinition& table_child,
+ const
ColumnDefinition& file_parent,
+
TableColumnMappingMode mode,
+ size_t
table_child_idx,
+ bool
allow_ordinal_fallback) {
+ const auto file_parent_type =
remove_nullable(file_parent.type)->get_primitive_type();
+ switch (file_parent_type) {
+ case TYPE_ARRAY:
+ DORIS_CHECK(file_parent.children.size() == 1);
+ return &file_parent.children[0];
+ case TYPE_MAP:
+ DORIS_CHECK(file_parent.children.size() == 2);
+ if (table_child.name == "key") {
+ return &file_parent.children[0];
+ }
+ if (table_child.name == "value") {
+ return &file_parent.children[1];
+ }
+ if (table_child.local_id == 0 || table_child.local_id == 1) {
+ return &file_parent.children[table_child.local_id];
+ }
+ return nullptr;
+ default:
+ // Hive BY_INDEX is a top-level column matching rule. Once a complex
root is selected by
+ // file position, nested struct children follow Hive reader's
historical name matching
+ // semantics; their integer identifiers can be field ids, not file
positions.
+ const auto nested_mode =
+ mode == TableColumnMappingMode::BY_INDEX ?
TableColumnMappingMode::BY_NAME : mode;
+ if (const auto* file_child =
+ matcher_for_mode(nested_mode).find(table_child,
file_parent.children);
+ file_child != nullptr) {
+ return file_child;
+ }
+ if (allow_ordinal_fallback && mode ==
TableColumnMappingMode::BY_FIELD_ID &&
+ !table_child.has_identifier_field_id()) {
+ // Synthetic children are derived from the table DataType when
nested ColumnDefinition
+ // metadata has been pruned away. They do not carry Iceberg field
ids, so try a name
+ // match before falling back to ordinal order. Example:
+ // table value type: Struct(age, full_name, gender)
+ // old file value: Struct(name, age)
+ // Name matching keeps `age -> age`; the later unused-child
fallback can then map the
+ // renamed `full_name -> name` instead of consuming `age` twice.
+ if (const auto* file_child = NameMatcher().find(table_child,
file_parent.children);
+ file_child != nullptr) {
+ return file_child;
+ }
+ }
+ // Some callers only carry the full complex DataType for a projected
table column, without
+ // expanded nested ColumnDefinitions. In that case we can still
preserve full materialization
+ // by walking table/file struct fields by ordinal. This is a fallback
only: explicit
+ // ColumnDefinition children keep using the requested table-format
matching rule, which is
+ // required for precise schema evolution.
+ if (allow_ordinal_fallback && table_child_idx <
file_parent.children.size()) {
+ return &file_parent.children[table_child_idx];
+ }
+ return nullptr;
+ }
+}
+
+static ColumnDefinition synthetic_child_definition(const std::string& name,
DataTypePtr type,
+ int32_t local_id) {
+ ColumnDefinition child;
+ child.identifier = Field::create_field<TYPE_STRING>(name);
+ child.local_id = local_id;
+ child.name = name;
+ child.type = std::move(type);
+ return child;
+}
+
+static std::vector<ColumnDefinition> synthesize_complex_children_from_type(
+ const DataTypePtr& type) {
+ std::vector<ColumnDefinition> children;
+ if (type == nullptr) {
+ return children;
+ }
+ const auto nested_type = remove_nullable(type);
+ switch (nested_type->get_primitive_type()) {
+ case TYPE_ARRAY: {
+ const auto* array_type = assert_cast<const
DataTypeArray*>(nested_type.get());
+ children.push_back(synthetic_child_definition("element",
array_type->get_nested_type(), 0));
+ break;
+ }
+ case TYPE_MAP: {
+ const auto* map_type = assert_cast<const
DataTypeMap*>(nested_type.get());
+ children.push_back(synthetic_child_definition("key",
map_type->get_key_type(), 0));
+ children.push_back(synthetic_child_definition("value",
map_type->get_value_type(), 1));
+ break;
+ }
+ case TYPE_STRUCT: {
+ const auto* struct_type = assert_cast<const
DataTypeStruct*>(nested_type.get());
+ children.reserve(struct_type->get_elements().size());
+ for (size_t idx = 0; idx < struct_type->get_elements().size(); ++idx) {
+
children.push_back(synthetic_child_definition(struct_type->get_element_name(idx),
+
struct_type->get_element(idx),
+
cast_set<int32_t>(idx)));
+ }
+ break;
+ }
+ default:
+ break;
+ }
+ return children;
+}
+
+static bool has_table_child_named(const std::vector<ColumnDefinition>&
children,
+ std::string_view name) {
+ return std::ranges::any_of(children, [&](const ColumnDefinition& child) {
+ return std::string_view(child.name) == name;
+ });
+}
+
+static void complete_required_complex_children_from_type(const DataTypePtr&
type,
+
std::vector<ColumnDefinition>* children) {
+ DORIS_CHECK(children != nullptr);
+ if (type == nullptr) {
+ return;
+ }
+ const auto nested_type = remove_nullable(type);
+ switch (nested_type->get_primitive_type()) {
+ case TYPE_MAP: {
+ const auto* map_type = assert_cast<const
DataTypeMap*>(nested_type.get());
+ // MAP key/value are structural children, not independently
materializable table fields.
+ // A key-only projection can still be attached to a whole-map output
root, for example:
+ // SELECT * FROM t WHERE ARRAY_CONTAINS(MAP_KEYS(new_map_column),
'person5')
+ //
+ // In that shape the scanner keeps the value stream readable, but the
table projection can
+ // carry only the key child. Add the missing value child so recursive
mapping can evolve the
+ // value type instead of letting TableReader cast old/new value
structs directly.
+ if (has_table_child_named(*children, "key") &&
!has_table_child_named(*children, "value")) {
+ children->push_back(synthetic_child_definition("value",
map_type->get_value_type(), 1));
+ }
+ break;
+ }
+ case TYPE_ARRAY:
+ // ARRAY has only one required structural child (`element`), so a
non-empty projection is
+ // already rooted at the element path.
+ break;
+ case TYPE_STRUCT:
+ // STRUCT children are real fields and must remain prunable.
Completing missing struct
+ // fields here would turn `SELECT s.a` into a full-struct read and
undo nested projection.
+ break;
+ default:
+ break;
+ }
+}
+
+static Status validate_file_schema_children(const ColumnDefinition&
file_field) {
+ if (file_field.type == nullptr) {
+ return Status::InternalError("File column '{}' has null type",
file_field.name);
+ }
+ const auto nested_type = remove_nullable(file_field.type);
+ size_t expected_children = 0;
+ bool complex_with_fixed_children = true;
+ switch (nested_type->get_primitive_type()) {
+ case TYPE_ARRAY:
+ expected_children = 1;
+ break;
+ case TYPE_MAP:
+ expected_children = 2;
+ break;
+ case TYPE_STRUCT:
+ expected_children =
+ assert_cast<const
DataTypeStruct*>(nested_type.get())->get_elements().size();
+ break;
+ default:
+ complex_with_fixed_children = false;
+ break;
+ }
+ if (!complex_with_fixed_children || file_field.children.size() ==
expected_children) {
+ return Status::OK();
+ }
+ return Status::InternalError(
+ "Malformed complex file schema for column '{}': type={},
expected_children={}, "
+ "actual_children={}",
+ file_field.name, file_field.type->get_name(), expected_children,
+ file_field.children.size());
+}
+
+static bool has_projected_file_children(const ColumnMapping& mapping) {
+ if (mapping.original_file_children.empty() ||
mapping.projected_file_children.empty()) {
+ return false;
+ }
+ if (mapping.original_file_children.size() !=
mapping.projected_file_children.size()) {
+ return true;
+ }
+ for (size_t idx = 0; idx < mapping.original_file_children.size(); ++idx) {
+ if (mapping.original_file_children[idx].file_local_id() !=
+ mapping.projected_file_children[idx].file_local_id()) {
+ return true;
+ }
+ }
+ return false;
+}
+
+static bool needs_nested_file_projection(const ColumnMapping& mapping) {
+ if (has_projected_file_children(mapping)) {
+ // Return True if the projected child column is missing / re-ordered
+ return true;
+ }
+ return std::ranges::any_of(mapping.child_mappings, [](const ColumnMapping&
child_mapping) {
+ return needs_nested_file_projection(child_mapping);
+ });
+}
+
+static Status build_complex_projection(const ColumnMapping& mapping,
LocalColumnIndex* projection);
+
+// Build the projected file children/type according to the pruned complex
projection. For example,
+// if we have a struct column `s` with children `id` and `name`, and the
projection only keeps
+// `s.name`, then the file reader should expose `STRUCT<name ...>`.
+static Status rebuild_projected_file_children_and_type(
+ const DataTypePtr& file_type, const std::vector<ColumnDefinition>&
original_file_children,
+ const std::vector<ColumnMapping>& child_mappings,
+ std::vector<ColumnDefinition>* projected_file_children, DataTypePtr*
projected_type) {
+ DORIS_CHECK(file_type != nullptr);
+ DORIS_CHECK(projected_file_children != nullptr);
+ DORIS_CHECK(projected_type != nullptr);
+ ColumnDefinition field;
+ field.type = file_type;
+ field.children = original_file_children;
+ LocalColumnIndex projection = LocalColumnIndex::partial_local(-1);
+ projection.children.reserve(child_mappings.size());
+ for (const auto* child_mapping :
present_child_mappings_in_file_order(child_mappings)) {
+ DORIS_CHECK(child_mapping->file_local_id.has_value());
+ LocalColumnIndex child_projection;
+ RETURN_IF_ERROR(build_complex_projection(*child_mapping,
&child_projection));
+ projection.children.push_back(std::move(child_projection));
+ }
+
+ ColumnDefinition projected_field;
+ RETURN_IF_ERROR(project_column_definition(field, projection,
&projected_field));
+ *projected_file_children = std::move(projected_field.children);
+ *projected_type = std::move(projected_field.type);
+ return Status::OK();
+}
+
+// Build the complex column projection according to the ColumnMapping which is
re-ordered by the
+// file-schema's order.
+//
+// For MAP, a partial projection represents value-subtree pruning only. The
key child is not a
+// projected output shape; file readers still read full keys to construct
ColumnMap offsets and keep
+// key semantics unchanged. If a caller tries to project only/prune the key
child, the common schema
+// projection helper rejects it.
+static Status build_complex_projection(const ColumnMapping& mapping,
LocalColumnIndex* projection) {
+ if (projection == nullptr) {
+ return Status::InvalidArgument("projection is null");
+ }
+ DORIS_CHECK(mapping.file_local_id.has_value());
+ *projection = LocalColumnIndex::local(*mapping.file_local_id);
+ projection->project_all_children = mapping.child_mappings.empty();
+ projection->children.clear();
+ const auto present_children =
present_child_mappings_in_file_order(mapping.child_mappings);
+ if (!projection->project_all_children && present_children.empty()) {
+ // All requested table children under this complex node are
missing/default-only. The file
+ // reader cannot expose an empty complex projection, but TableReader
can still rematerialize
+ // the table shape from a full file subtree and fill the missing
children with defaults.
+ projection->project_all_children = true;
+ return Status::OK();
+ }
+ for (const auto* child_mapping : present_children) {
+ LocalColumnIndex child_projection;
+ RETURN_IF_ERROR(build_complex_projection(*child_mapping,
&child_projection));
+ projection->children.push_back(std::move(child_projection));
+ }
+ if (!projection->project_all_children && projection->children.empty()) {
+ return Status::NotSupported("Projection for complex column {} contains
no file children",
+ mapping.file_column_name);
+ }
+ return Status::OK();
+}
+
+using FilterProjectionMap = std::map<LocalColumnId, LocalColumnIndex>;
+
+// Update the mapping's file type according to the projection, and determine
whether the projection
+// is trivial (i.e. the projected file type is the same as the table type, so
no need to
+// rematerialize the complex value back to table layout after reading from
file).
+static Status apply_projection_to_mapping_file_type(const LocalColumnIndex&
projection,
+ ColumnMapping* mapping) {
+ DORIS_CHECK(mapping != nullptr);
+ if (mapping->original_file_type == nullptr) {
+ mapping->original_file_type = mapping->file_type;
+ }
+ if (mapping->original_file_type == nullptr ||
+
!is_complex_type(remove_nullable(mapping->original_file_type)->get_primitive_type()))
{
+ return Status::OK();
+ }
+ ColumnDefinition field;
+ field.type = mapping->original_file_type;
+ field.children = mapping->original_file_children;
+ ColumnDefinition projected_field;
+ RETURN_IF_ERROR(project_column_definition(field, projection,
&projected_field));
+ mapping->file_type = std::move(projected_field.type);
+ mapping->projected_file_children = std::move(projected_field.children);
+ mapping->is_trivial = mapping_can_use_file_column_directly(*mapping);
+ return Status::OK();
+}
+
+static Status merge_filter_projection(const FilterProjectionMap*
filter_projections,
+ LocalColumnIndex* projection) {
+ DORIS_CHECK(projection != nullptr);
+ if (filter_projections == nullptr) {
+ return Status::OK();
+ }
+ const auto filter_projection_it =
filter_projections->find(projection->column_id());
+ if (filter_projection_it == filter_projections->end()) {
+ return Status::OK();
+ }
+ // Merge predicate-only nested paths into the root projection that is
about to be scanned.
+ // Example: `SELECT s.a WHERE s.b > 1` first builds the output projection
`s -> a` from
+ // ColumnMapping, while build_nested_struct_filter_projection_map()
records `s -> b`. This merge
+ // produces one file scan projection `s -> a,b`.
+ RETURN_IF_ERROR(merge_local_column_index(projection,
filter_projection_it->second));
+ return Status::OK();
+}
+
+static bool table_root_is_map(const ColumnMapping& mapping) {
+ if (mapping.table_type == nullptr) {
+ return false;
+ }
+ return remove_nullable(mapping.table_type)->get_primitive_type() ==
TYPE_MAP;
+}
+
+static Status add_scan_column(FileScanRequest* file_request, ColumnMapping*
mapping,
+ bool is_predicate_column, bool
force_full_complex_scan_projection,
+ const FilterProjectionMap* filter_projections =
nullptr) {
+ const auto file_column_id = LocalColumnId(mapping->file_local_id.value());
+ LocalColumnIndex projection = LocalColumnIndex::top_level(file_column_id);
+ // Columnar readers can turn a complex mapping into a nested file
projection, but
+ // row-oriented readers must scan the full top-level complex field because
all children are
+ // encoded in the same text cell.
+ if (!force_full_complex_scan_projection &&
needs_nested_file_projection(*mapping)) {
+ RETURN_IF_ERROR(build_complex_projection(*mapping, &projection));
+ }
+ if (is_predicate_column && !force_full_complex_scan_projection) {
+ DCHECK(filter_projections != nullptr);
+ // If a projected complex root is also used by a predicate, rebuild
the predicate scan
+ // projection from the output mapping before merging predicate-only
children. For
+ // `SELECT s.a WHERE s.b > 1`, build_complex_projection() produces `s
-> a` and
+ // merge_filter_projection() adds `s -> b`, so the predicate column
reads both children.
+ RETURN_IF_ERROR(merge_filter_projection(filter_projections,
&projection));
+ }
+ FileScanRequestBuilder builder(file_request);
+ if (is_predicate_column) {
+ return builder.add_predicate_column(std::move(projection));
+ }
+ return builder.add_non_predicate_column(std::move(projection));
+}
+
+static const LocalColumnIndex* find_scan_projection(
+ const std::vector<LocalColumnIndex>& scan_columns, LocalColumnId
file_column_id) {
+ const auto projection_it =
+ std::ranges::find_if(scan_columns, [&](const LocalColumnIndex&
projection) {
+ return projection.column_id() == file_column_id;
+ });
+ return projection_it == scan_columns.end() ? nullptr : &*projection_it;
+}
+
+// Apply the final scan projection of one root file column back to its
ColumnMapping. This updates
+// mapping.file_type/projected_file_children from the original file schema to
the exact shape that
+// FileReader will return.
+//
+// Example: for `SELECT s.a WHERE s.b > 1`, add_scan_column() keeps only one
predicate scan
+// projection `s -> a,b`. Applying that projection changes the mapping's file
type from the full
+// file struct `s<a,b,c>` to the projected file struct `s<a,b>`, so later
filter rewrite and
+// TableReader final materialization use the same column shape as the
file-local block.
+static Status apply_scan_projection_to_mapping_file_type(const
FileScanRequest& file_request,
+ ColumnMapping*
mapping) {
+ DORIS_CHECK(mapping != nullptr);
+ DORIS_CHECK(mapping->file_local_id.has_value());
+ const auto file_column_id = LocalColumnId(*mapping->file_local_id);
+ // Predicate columns are the actual scan projection when a column is used
by row-level filters:
+ // add_scan_column() removes the duplicate non-predicate projection in
that case.
+ const auto* projection =
find_scan_projection(file_request.predicate_columns, file_column_id);
+ if (projection == nullptr) {
+ projection = find_scan_projection(file_request.non_predicate_columns,
file_column_id);
+ }
+ DORIS_CHECK(projection != nullptr);
+ return apply_projection_to_mapping_file_type(*projection, mapping);
+}
+
+// Build extra scan projections required only by row-level filters on nested
struct children.
+//
+// Example: for `SELECT s.a FROM t WHERE s.b.c > 1`, the output projection may
only contain `s.a`,
+// but the file reader must also read `s.b.c` to evaluate the predicate. This
function collects the
+// table-side filter path, resolves it through ColumnMapping first, and
records the corresponding
+// file-side projection in filter_projections. This keeps renamed fields
consistent across the scan
+// projection, row-level conjunct rewrite, and nested predicate pruning.
Example:
+// table filter path: s -> renamed_b -> c
+// old file path: s -> b -> c
+// recorded path: s -> b -> c
+// When add_scan_column() adds the same root as a predicate column, it
rebuilds that root from the
+// output mapping, merges this filter-only projection into it, and removes the
duplicate
+// non-predicate root entry.
+static Status build_nested_struct_filter_projection_map(
+ const std::vector<TableFilter>& table_filters, const
std::vector<ColumnMapping>& mappings,
+ FilterProjectionMap* filter_projections) {
+ DORIS_CHECK(filter_projections != nullptr);
+ filter_projections->clear();
+ for (const auto& table_filter : table_filters) {
+ if (table_filter.conjunct == nullptr) {
+ continue;
+ }
+ // Collect all nested struct paths in the table filter. For example,
for
+ // `s.id > 5 AND element_at(s, 'renamed_name') = 'abc'`, collect the
table paths
+ // `s -> id` and `s -> renamed_name`, then resolve each one to its
file-side projection.
+ std::vector<NestedStructPath> paths;
+ collect_nested_struct_paths(table_filter.conjunct->root(), &paths);
+ for (const auto& path : paths) {
+ auto mapping_it = std::ranges::find_if(mappings, [&](const
ColumnMapping& mapping) {
+ return mapping.global_index == path.root_global_index;
+ });
+ if (mapping_it == mappings.end() ||
!mapping_it->file_local_id.has_value() ||
+ path.selectors.empty()) {
+ continue;
+ }
+
+ ResolvedNestedStructPath resolved;
+ LocalColumnIndex root_projection;
+ if (!resolve_nested_struct_path_for_file(path, mappings,
&resolved)) {
+ if (!table_root_is_map(*mapping_it)) {
+ continue;
+ }
+ // Direct map value filters such as `m.value.a > 1` need the
value leaf for row
+ // evaluation even when the query only projects another value
child. This is only a
+ // scan projection fallback; complex map/array expressions are
still not rewritten
+ // into file-local conjuncts.
+ LocalColumnIndex child_projection;
+ RETURN_IF_ERROR(build_file_child_projection_from_schema(
+ mapping_it->original_file_children, path.selectors,
&child_projection));
+ if (child_projection.local_id() < 0) {
+ continue;
+ }
+ root_projection =
LocalColumnIndex::partial_local(*mapping_it->file_local_id);
+
root_projection.children.push_back(std::move(child_projection));
+ } else {
+ root_projection = std::move(resolved.file_projection);
+ }
+ auto filter_projection_it =
filter_projections->find(root_projection.column_id());
+ if (filter_projection_it == filter_projections->end()) {
+ filter_projections->emplace(root_projection.column_id(),
+ std::move(root_projection));
+ continue;
+ }
+ RETURN_IF_ERROR(
+ merge_local_column_index(&filter_projection_it->second,
root_projection));
+ }
+ }
+ return Status::OK();
+}
+
+static void rebuild_projection(ColumnMapping* mapping, LocalIndex
block_position) {
+ DORIS_CHECK(mapping->file_local_id.has_value());
+ if (mapping->is_trivial || needs_complex_rematerialize(*mapping)) {
+ mapping->projection =
VExprContext::create_shared(VSlotRef::create_shared(
+ cast_set<int>(block_position.value()),
cast_set<int>(block_position.value()), -1,
+ mapping->file_type, mapping->file_column_name));
+ return;
+ }
+
+ auto expr = Cast::create_shared(mapping->table_type);
+
expr->add_child(VSlotRef::create_shared(cast_set<int>(block_position.value()),
+
cast_set<int>(block_position.value()), -1,
+ mapping->file_type,
mapping->file_column_name));
+ mapping->projection = VExprContext::create_shared(expr);
+}
+
+// Build file slot rewrite info from the localized filter targets. Only local
targets can enter
+// file-reader expressions; constant and unset targets stay above the file
reader.
+static std::map<GlobalIndex, FileSlotRewriteInfo> build_file_slot_rewrite_map(
+ const std::vector<ColumnMapping>& mappings,
+ const std::map<GlobalIndex, FilterEntry>& filter_entries) {
+ std::map<GlobalIndex, FileSlotRewriteInfo> global_to_file_slot;
+ for (const auto& mapping : mappings) {
+ const auto entry_it = filter_entries.find(mapping.global_index);
+ if (entry_it == filter_entries.end() || !entry_it->second.is_local()) {
+ continue;
+ }
+ DORIS_CHECK(mapping.file_local_id.has_value());
+ global_to_file_slot.emplace(
+ mapping.global_index,
+ FileSlotRewriteInfo {.block_position =
entry_it->second.local_index().value(),
+ .file_type = mapping.file_type,
+ .table_type = mapping.table_type,
+ .file_column_name =
mapping.file_column_name});
+ }
+ return global_to_file_slot;
+}
+
+Status TableColumnMapper::_create_by_index_mapping(const ColumnDefinition&
table_column,
+ const
std::vector<ColumnDefinition>& file_schema,
+ ColumnMapping* mapping) {
+ DORIS_CHECK(mapping != nullptr);
+ DORIS_CHECK(!table_column.is_partition_key);
+
+ // Key contract: in BY_INDEX mode, `ColumnDefinition::identifier` TYPE_INT
is interpreted as the
+ // 0-based position of this column inside `file_schema`. FE writes the
physical file position
+ // of each non-partition projected column into that identifier. This
interpretation allows:
+ // - sparse projection: read only a subset of file columns (for example
only `_col2`
+ // and `_col4`);
+ // - column reordering: table column order differs from file column
order;
+ // - no many-to-one mapping: FE must guarantee that each file position
is referenced by at
+ // most one table column.
+ const auto file_index = table_column.get_identifier_position();
+
+ // Case A: file_index is in range, so build a direct positional mapping.
+ // The file column name (for example `_col0`) is intentionally ignored
here.
+ if (file_index >= 0 && static_cast<size_t>(file_index) <
file_schema.size()) {
+ return _create_direct_mapping(table_column,
file_schema[static_cast<size_t>(file_index)],
+ mapping);
+ }
+
+ // Case B: file_index is out of range, which means the file does not
contain this column.
+ // Route it through the missing-column path used by schema evolution.
+ if (table_column.default_expr != nullptr) {
+ _set_constant_mapping(mapping, table_column.default_expr);
+ return Status::OK();
+ }
+ // Keep the mapping empty (`file_local_id` remains `nullopt`) and let the
upper finalize
+ // stage fill NULL/default values.
+ return Status::OK();
+}
+
+void TableColumnMapper::_set_constant_mapping(ColumnMapping* mapping,
VExprContextSPtr expr) {
+ DORIS_CHECK(mapping != nullptr);
+ DORIS_CHECK(expr != nullptr);
+ mapping->default_expr = std::move(expr);
+ mapping->constant_index = _constant_map.add(ConstantEntry {
+ .global_index = mapping->global_index,
+ .expr = mapping->default_expr,
+ .type = mapping->table_type,
+ });
+ mapping->filter_conversion = FilterConversionType::CONSTANT;
+}
+
+Status TableColumnMapper::_create_mapping_for_column(const ColumnDefinition&
table_column,
+ GlobalIndex global_index,
+ ColumnMapping* mapping) {
+ DORIS_CHECK(mapping != nullptr);
+ *mapping = ColumnMapping {};
+ mapping->global_index = global_index;
+ mapping->table_column_name = table_column.name;
+ mapping->table_type = table_column.type;
+ const auto row_lineage_type =
row_lineage_virtual_column_type(table_column, _options.mode);
+ if (const auto* partition_value = find_partition_value(table_column,
_partition_values);
+ table_column.is_partition_key && partition_value != nullptr) {
+ // Partition values are split constants and must take precedence over
defaults.
+ _set_constant_mapping(mapping,
VExprContext::create_shared(VLiteral::create_shared(
+ mapping->table_type,
*partition_value)));
+ } else if (_options.mode == TableColumnMappingMode::BY_INDEX &&
+ !table_column.is_partition_key &&
table_column.has_identifier_field_id()) {
+ // BY_INDEX interprets ColumnDefinition::identifier as physical file
position.
+ RETURN_IF_ERROR(_create_by_index_mapping(table_column, _file_schema,
mapping));
+ } else if (const auto* file_field = _find_file_field(table_column,
_file_schema)) {
+ // Normal physical file column mapping.
+ RETURN_IF_ERROR(_create_direct_mapping(table_column, *file_field,
mapping));
+ if (row_lineage_type != TableVirtualColumnType::INVALID) {
+ // Iceberg v3 rewritten files may physically contain row lineage
metadata fields.
+ // File non-null values must be preserved, while file NULLs still
inherit from data file
+ // metadata in IcebergTableReader. Therefore the mapping has a
real file source plus a
+ // virtual post-materialization step, and filters must wait for
finalize output.
+ mapping->virtual_column_type = row_lineage_type;
+ mapping->filter_conversion = FilterConversionType::FINALIZE_ONLY;
+ }
+ } else if (row_lineage_type != TableVirtualColumnType::INVALID) {
+ // Iceberg row lineage metadata fields are optional in data files.
Missing fields are exposed
+ // as all-NULL table columns first; IcebergTableReader fills inherited
values only when the
+ // split carries first_row_id / last_updated_sequence_number metadata.
+ // FE may attach a default_expr to these hidden metadata columns, but
the Iceberg v3
+ // inheritance rule must take precedence over the generic
missing-column default path.
+ mapping->virtual_column_type = row_lineage_type;
+ } else if (table_column.name == BeConsts::ICEBERG_ROWID_COL) {
+ // Doris internal Iceberg row locator is never a physical Iceberg data
column. It is built
+ // from file path, row position and partition metadata for
delete/update/merge.
+ mapping->virtual_column_type = TableVirtualColumnType::ICEBERG_ROWID;
+ } else if (table_column.default_expr != nullptr) {
+ // Missing schema-evolution column with an explicit default expression.
+ _set_constant_mapping(mapping, table_column.default_expr);
+ } else {
+ if (table_column.is_partition_key) {
+ return Status::InvalidArgument(
+ "Table column '{}' (global_index={}) does not have a
matching partition value",
+ table_column.name, mapping->global_index.value());
+ }
+ }
+ return Status::OK();
+}
+
+Status TableColumnMapper::_create_hidden_filter_mapping(const
ColumnDefinition& table_column,
+ GlobalIndex
global_index,
+ ColumnMapping*
mapping) {
+ auto status = _create_mapping_for_column(table_column, global_index,
mapping);
+ if (mapping->file_local_id.has_value() ||
mapping->constant_index.has_value() ||
+ mapping->virtual_column_type != TableVirtualColumnType::INVALID) {
+ return Status::OK();
+ }
+ if (_options.mode == TableColumnMappingMode::BY_NAME) {
+ return status;
+ }
+
+ // Predicate-only slot refs carry the table name/type but do not carry the
table-format field
+ // id used by BY_FIELD_ID or the file position used by BY_INDEX. Use a
name fallback only for
+ // hidden filter localization; projected columns still obey the requested
mapping mode.
+ const auto* file_field =
+
matcher_for_mode(TableColumnMappingMode::BY_NAME).find(table_column,
_file_schema);
Review Comment:
Here the fallback maps predicate-only slots by name even when the table is
using BY_FIELD_ID or BY_INDEX mapping. For schema-evolved formats, a
filter-only table column can share a name with an old file field while having a
different field id/position; V2 then adds this hidden mapping as a predicate
column and Parquet applies it before output columns are read, so rows filtered
against the wrong physical field are lost before the normal table-level
conjunct can correct them. Please avoid localizing hidden filters unless the
slot can be resolved in the active mapping mode, or carry the field id/position
into the hidden `ColumnDefinition`.
--
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]
