Coverage Report

Created: 2026-08-06 13:30

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/format_v2/column_mapper.cpp
Line
Count
Source
1
// Licensed to the Apache Software Foundation (ASF) under one
2
// or more contributor license agreements.  See the NOTICE file
3
// distributed with this work for additional information
4
// regarding copyright ownership.  The ASF licenses this file
5
// to you under the Apache License, Version 2.0 (the
6
// "License"); you may not use this file except in compliance
7
// with the License.  You may obtain a copy of the License at
8
//
9
//   http://www.apache.org/licenses/LICENSE-2.0
10
//
11
// Unless required by applicable law or agreed to in writing,
12
// software distributed under the License is distributed on an
13
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14
// KIND, either express or implied.  See the License for the
15
// specific language governing permissions and limitations
16
// under the License.
17
18
#include "format_v2/column_mapper.h"
19
20
#include <algorithm>
21
#include <cstddef>
22
#include <memory>
23
#include <optional>
24
#include <set>
25
#include <sstream>
26
#include <string_view>
27
#include <utility>
28
#include <vector>
29
30
#include "common/consts.h"
31
#include "common/exception.h"
32
#include "common/status.h"
33
#include "core/data_type/convert_field_to_type.h"
34
#include "core/data_type/data_type_array.h"
35
#include "core/data_type/data_type_map.h"
36
#include "core/data_type/data_type_nullable.h"
37
#include "core/data_type/data_type_string.h"
38
#include "core/data_type/data_type_struct.h"
39
#include "core/data_type/primitive_type.h"
40
#include "exprs/runtime_filter_expr.h"
41
#include "exprs/short_circuit_evaluation_expr.h"
42
#include "exprs/vcase_expr.h"
43
#include "exprs/vcast_expr.h"
44
#include "exprs/vcondition_expr.h"
45
#include "exprs/vectorized_fn_call.h"
46
#include "exprs/vexpr_context.h"
47
#include "exprs/vin_predicate.h"
48
#include "exprs/vliteral.h"
49
#include "format_v2/column_mapper_nested.h"
50
#include "format_v2/expr/cast.h"
51
#include "format_v2/file_reader.h"
52
#include "format_v2/schema_projection.h"
53
#include "format_v2/table_reader.h"
54
#include "gen_cpp/Exprs_types.h"
55
56
namespace doris::format {
57
58
namespace {
59
60
11
bool has_shared_descendant_field_id(const ColumnDefinition& table, const ColumnDefinition& file) {
61
11
    const auto& table_children =
62
11
            table.identity_children.empty() ? table.children : table.identity_children;
63
11
    for (const auto& table_child : table_children) {
64
11
        if (!table_child.has_identifier_field_id()) {
65
0
            continue;
66
0
        }
67
11
        const auto file_child =
68
11
                std::ranges::find_if(file.children, [&](const ColumnDefinition& candidate) {
69
11
                    return candidate.has_identifier_field_id() &&
70
11
                           candidate.get_identifier_field_id() ==
71
10
                                   table_child.get_identifier_field_id();
72
11
                });
73
11
        if (file_child != file.children.end() ||
74
11
            std::ranges::any_of(file.children, [&](const ColumnDefinition& candidate) {
75
2
                return has_shared_descendant_field_id(table_child, candidate);
76
10
            })) {
77
10
            return true;
78
10
        }
79
11
    }
80
1
    return false;
81
11
}
82
83
16
std::string mapping_mode_to_string(TableColumnMappingMode mode) {
84
16
    switch (mode) {
85
1
    case TableColumnMappingMode::BY_FIELD_ID:
86
1
        return "BY_FIELD_ID";
87
14
    case TableColumnMappingMode::BY_NAME:
88
14
        return "BY_NAME";
89
1
    case TableColumnMappingMode::BY_INDEX:
90
1
        return "BY_INDEX";
91
16
    }
92
0
    return "UNKNOWN";
93
16
}
94
95
13.2M
bool column_has_name(const ColumnDefinition& column, const std::string& name) {
96
13.2M
    if (to_lower(column.name) == to_lower(name)) {
97
302k
        return true;
98
302k
    }
99
12.9M
    if (column.has_identifier_name() && to_lower(column.get_identifier_name()) == to_lower(name)) {
100
0
        return true;
101
0
    }
102
12.9M
    return std::ranges::any_of(column.name_mapping, [&](const std::string& alias) {
103
146
        return to_lower(alias) == to_lower(name);
104
146
    });
105
12.9M
}
106
107
6.90M
bool column_names_match(const ColumnDefinition& lhs, const ColumnDefinition& rhs) {
108
6.90M
    if (!lhs.has_name_mapping) {
109
6.90M
        if (column_has_name(rhs, lhs.name)) {
110
302k
            return true;
111
302k
        }
112
6.60M
        if (lhs.has_identifier_name() && column_has_name(rhs, lhs.get_identifier_name())) {
113
1
            return true;
114
1
        }
115
6.60M
    }
116
    // Explicit Iceberg name mapping is authoritative: an empty alias list represents a field that
117
    // did not exist in the imported file, so only transported aliases may match.
118
6.60M
    return std::ranges::any_of(lhs.name_mapping, [&](const std::string& alias) {
119
32
        return column_has_name(rhs, alias);
120
32
    });
121
6.90M
}
122
123
class ColumnMatcher {
124
public:
125
3
    virtual ~ColumnMatcher() = default;
126
    virtual const ColumnDefinition* find(
127
            const ColumnDefinition& table_column,
128
            const std::vector<ColumnDefinition>& file_schema) const = 0;
129
};
130
131
class FieldIdMatcher final : public ColumnMatcher {
132
public:
133
    const ColumnDefinition* find(const ColumnDefinition& table_column,
134
275k
                                 const std::vector<ColumnDefinition>& file_schema) const override {
135
275k
        if (!table_column.has_identifier_field_id()) {
136
565
            return nullptr;
137
565
        }
138
274k
        const auto field_id = table_column.get_identifier_field_id();
139
2.25M
        const auto field_it = std::ranges::find_if(file_schema, [&](const ColumnDefinition& field) {
140
2.25M
            return field.has_identifier_field_id() && field.get_identifier_field_id() == field_id;
141
2.25M
        });
142
274k
        return field_it == file_schema.end() ? nullptr : &*field_it;
143
275k
    }
144
};
145
146
class NameMatcher final : public ColumnMatcher {
147
public:
148
    const ColumnDefinition* find(const ColumnDefinition& table_column,
149
303k
                                 const std::vector<ColumnDefinition>& file_schema) const override {
150
6.91M
        const auto field_it = std::ranges::find_if(file_schema, [&](const ColumnDefinition& field) {
151
6.91M
            return column_names_match(table_column, field);
152
6.91M
        });
153
303k
        return field_it == file_schema.end() ? nullptr : &*field_it;
154
303k
    }
155
};
156
157
class PositionMatcher final : public ColumnMatcher {
158
public:
159
    const ColumnDefinition* find(const ColumnDefinition& table_column,
160
2
                                 const std::vector<ColumnDefinition>& file_schema) const override {
161
2
        if (!table_column.has_identifier_field_id()) {
162
2
            return nullptr;
163
2
        }
164
0
        const auto position = table_column.get_identifier_position();
165
0
        if (position < 0 || static_cast<size_t>(position) >= file_schema.size()) {
166
0
            return nullptr;
167
0
        }
168
0
        return &file_schema[static_cast<size_t>(position)];
169
0
    }
170
};
171
172
578k
const ColumnMatcher& matcher_for_mode(TableColumnMappingMode mode) {
173
578k
    static const FieldIdMatcher field_id_matcher;
174
578k
    static const NameMatcher name_matcher;
175
578k
    static const PositionMatcher position_matcher;
176
578k
    switch (mode) {
177
275k
    case TableColumnMappingMode::BY_FIELD_ID:
178
275k
        return field_id_matcher;
179
303k
    case TableColumnMappingMode::BY_NAME:
180
303k
        return name_matcher;
181
2
    case TableColumnMappingMode::BY_INDEX:
182
2
        return position_matcher;
183
578k
    }
184
0
    return field_id_matcher;
185
578k
}
186
187
12
std::string virtual_column_type_to_string(TableVirtualColumnType type) {
188
12
    switch (type) {
189
9
    case TableVirtualColumnType::INVALID:
190
9
        return "INVALID";
191
1
    case TableVirtualColumnType::ROW_ID:
192
1
        return "ROW_ID";
193
1
    case TableVirtualColumnType::LAST_UPDATED_SEQUENCE_NUMBER:
194
1
        return "LAST_UPDATED_SEQUENCE_NUMBER";
195
1
    case TableVirtualColumnType::ICEBERG_ROWID:
196
1
        return "ICEBERG_ROWID";
197
12
    }
198
0
    return "UNKNOWN";
199
12
}
200
201
12
std::string filter_conversion_type_to_string(FilterConversionType type) {
202
12
    switch (type) {
203
3
    case FilterConversionType::COPY_DIRECTLY:
204
3
        return "COPY_DIRECTLY";
205
1
    case FilterConversionType::CAST_FILTER:
206
1
        return "CAST_FILTER";
207
1
    case FilterConversionType::READER_EXPRESSION:
208
1
        return "READER_EXPRESSION";
209
6
    case FilterConversionType::FINALIZE_ONLY:
210
6
        return "FINALIZE_ONLY";
211
1
    case FilterConversionType::CONSTANT:
212
1
        return "CONSTANT";
213
12
    }
214
0
    return "UNKNOWN";
215
12
}
216
217
47
std::string data_type_debug_string(const DataTypePtr& type) {
218
47
    return type == nullptr ? "null" : type->get_name();
219
47
}
220
221
11
std::string field_debug_string(const Field& field) {
222
11
    std::ostringstream out;
223
11
    out << "Field{type=" << type_to_string(field.get_type()) << ", value=";
224
11
    switch (field.get_type()) {
225
0
    case TYPE_NULL:
226
0
        out << "null";
227
0
        break;
228
9
    case TYPE_INT:
229
9
        out << field.get<TYPE_INT>();
230
9
        break;
231
0
    case TYPE_BIGINT:
232
0
        out << field.get<TYPE_BIGINT>();
233
0
        break;
234
2
    case TYPE_STRING:
235
2
        out << field.get<TYPE_STRING>();
236
2
        break;
237
0
    default:
238
0
        out << field.to_debug_string(0);
239
0
        break;
240
11
    }
241
11
    out << "}";
242
11
    return out.str();
243
11
}
244
245
template <typename T, typename Formatter>
246
61
std::string join_debug_strings(const std::vector<T>& values, Formatter formatter) {
247
61
    std::ostringstream out;
248
61
    out << "[";
249
83
    for (size_t i = 0; i < values.size(); ++i) {
250
22
        if (i > 0) {
251
1
            out << ", ";
252
1
        }
253
22
        out << formatter(values[i]);
254
22
    }
255
61
    out << "]";
256
61
    return out.str();
257
61
}
column_mapper.cpp:_ZN5doris6format12_GLOBAL__N_118join_debug_stringsINSt7__cxx1112basic_stringIcSt11char_traitsIcESaIcEEEZNKS0_16ColumnDefinition12debug_stringEvE3$_0EES8_RKSt6vectorIT_SaISC_EET0_
Line
Count
Source
246
11
std::string join_debug_strings(const std::vector<T>& values, Formatter formatter) {
247
11
    std::ostringstream out;
248
11
    out << "[";
249
19
    for (size_t i = 0; i < values.size(); ++i) {
250
8
        if (i > 0) {
251
0
            out << ", ";
252
0
        }
253
8
        out << formatter(values[i]);
254
8
    }
255
11
    out << "]";
256
11
    return out.str();
257
11
}
column_mapper.cpp:_ZN5doris6format12_GLOBAL__N_118join_debug_stringsINS0_16ColumnDefinitionEZNKS3_12debug_stringB5cxx11EvE3$_1EENSt7__cxx1112basic_stringIcSt11char_traitsIcESaIcEEERKSt6vectorIT_SaISC_EET0_
Line
Count
Source
246
11
std::string join_debug_strings(const std::vector<T>& values, Formatter formatter) {
247
11
    std::ostringstream out;
248
11
    out << "[";
249
12
    for (size_t i = 0; i < values.size(); ++i) {
250
1
        if (i > 0) {
251
0
            out << ", ";
252
0
        }
253
1
        out << formatter(values[i]);
254
1
    }
255
11
    out << "]";
256
11
    return out.str();
257
11
}
column_mapper.cpp:_ZN5doris6format12_GLOBAL__N_118join_debug_stringsINS0_16ColumnDefinitionEZNKS3_12debug_stringB5cxx11EvE3$_2EENSt7__cxx1112basic_stringIcSt11char_traitsIcESaIcEEERKSt6vectorIT_SaISC_EET0_
Line
Count
Source
246
11
std::string join_debug_strings(const std::vector<T>& values, Formatter formatter) {
247
11
    std::ostringstream out;
248
11
    out << "[";
249
11
    for (size_t i = 0; i < values.size(); ++i) {
250
0
        if (i > 0) {
251
0
            out << ", ";
252
0
        }
253
0
        out << formatter(values[i]);
254
0
    }
255
11
    out << "]";
256
11
    return out.str();
257
11
}
column_mapper.cpp:_ZN5doris6format12_GLOBAL__N_118join_debug_stringsINS0_16LocalColumnIndexEZNKS3_12debug_stringB5cxx11EvE3$_0EENSt7__cxx1112basic_stringIcSt11char_traitsIcESaIcEEERKSt6vectorIT_SaISC_EET0_
Line
Count
Source
246
2
std::string join_debug_strings(const std::vector<T>& values, Formatter formatter) {
247
2
    std::ostringstream out;
248
2
    out << "[";
249
3
    for (size_t i = 0; i < values.size(); ++i) {
250
1
        if (i > 0) {
251
0
            out << ", ";
252
0
        }
253
1
        out << formatter(values[i]);
254
1
    }
255
2
    out << "]";
256
2
    return out.str();
257
2
}
column_mapper.cpp:_ZN5doris6format12_GLOBAL__N_118join_debug_stringsINS0_16ColumnDefinitionEZNKS0_13ColumnMapping12debug_stringB5cxx11EvE3$_0EENSt7__cxx1112basic_stringIcSt11char_traitsIcESaIcEEERKSt6vectorIT_SaISD_EET0_
Line
Count
Source
246
12
std::string join_debug_strings(const std::vector<T>& values, Formatter formatter) {
247
12
    std::ostringstream out;
248
12
    out << "[";
249
17
    for (size_t i = 0; i < values.size(); ++i) {
250
5
        if (i > 0) {
251
0
            out << ", ";
252
0
        }
253
5
        out << formatter(values[i]);
254
5
    }
255
12
    out << "]";
256
12
    return out.str();
257
12
}
column_mapper.cpp:_ZN5doris6format12_GLOBAL__N_118join_debug_stringsINS0_13ColumnMappingEZNKS3_12debug_stringB5cxx11EvE3$_1EENSt7__cxx1112basic_stringIcSt11char_traitsIcESaIcEEERKSt6vectorIT_SaISC_EET0_
Line
Count
Source
246
12
std::string join_debug_strings(const std::vector<T>& values, Formatter formatter) {
247
12
    std::ostringstream out;
248
12
    out << "[";
249
17
    for (size_t i = 0; i < values.size(); ++i) {
250
5
        if (i > 0) {
251
0
            out << ", ";
252
0
        }
253
5
        out << formatter(values[i]);
254
5
    }
255
12
    out << "]";
256
12
    return out.str();
257
12
}
column_mapper.cpp:_ZN5doris6format12_GLOBAL__N_118join_debug_stringsINS0_13ColumnMappingEZNKS0_17TableColumnMapper12debug_stringB5cxx11EvE3$_0EENSt7__cxx1112basic_stringIcSt11char_traitsIcESaIcEEERKSt6vectorIT_SaISD_EET0_
Line
Count
Source
246
1
std::string join_debug_strings(const std::vector<T>& values, Formatter formatter) {
247
1
    std::ostringstream out;
248
1
    out << "[";
249
3
    for (size_t i = 0; i < values.size(); ++i) {
250
2
        if (i > 0) {
251
1
            out << ", ";
252
1
        }
253
2
        out << formatter(values[i]);
254
2
    }
255
1
    out << "]";
256
1
    return out.str();
257
1
}
column_mapper.cpp:_ZN5doris6format12_GLOBAL__N_118join_debug_stringsINS0_13ColumnMappingEZNKS0_17TableColumnMapper12debug_stringB5cxx11EvE3$_1EENSt7__cxx1112basic_stringIcSt11char_traitsIcESaIcEEERKSt6vectorIT_SaISD_EET0_
Line
Count
Source
246
1
std::string join_debug_strings(const std::vector<T>& values, Formatter formatter) {
247
1
    std::ostringstream out;
248
1
    out << "[";
249
1
    for (size_t i = 0; i < values.size(); ++i) {
250
0
        if (i > 0) {
251
0
            out << ", ";
252
0
        }
253
0
        out << formatter(values[i]);
254
0
    }
255
1
    out << "]";
256
1
    return out.str();
257
1
}
258
259
} // namespace
260
261
const ColumnDefinition* find_column_by_name(const ColumnDefinition& table_column,
262
22
                                            const std::vector<ColumnDefinition>& file_schema) {
263
22
    return matcher_for_mode(TableColumnMappingMode::BY_NAME).find(table_column, file_schema);
264
22
}
265
266
const ColumnDefinition* find_column_by_field_id(const ColumnDefinition& table_column,
267
                                                const std::vector<ColumnDefinition>& file_schema,
268
231k
                                                bool allow_idless_complex_wrapper_projection) {
269
231k
    const auto* matched =
270
231k
            matcher_for_mode(TableColumnMappingMode::BY_FIELD_ID).find(table_column, file_schema);
271
231k
    if (matched != nullptr || !allow_idless_complex_wrapper_projection ||
272
231k
        table_column.children.empty()) {
273
231k
        return matched;
274
231k
    }
275
904
    const ColumnDefinition* wrapper = nullptr;
276
3.46k
    for (const auto& candidate : file_schema) {
277
3.46k
        if (candidate.has_identifier_field_id() || candidate.children.empty() ||
278
3.46k
            !has_shared_descendant_field_id(table_column, candidate)) {
279
3.45k
            continue;
280
3.45k
        }
281
9
        if (wrapper != nullptr) {
282
0
            return nullptr;
283
0
        }
284
9
        wrapper = &candidate;
285
9
    }
286
    // Iceberg Parquet's PruneColumns retains an ID-less complex wrapper when a nested field ID is
287
    // selected. Descendant IDs, not aliases, identify that wrapper; ambiguity remains unmapped.
288
904
    return wrapper;
289
904
}
290
291
const Field* find_partition_value(const ColumnDefinition& table_column,
292
545k
                                  const std::map<std::string, Field>& partition_values) {
293
804k
    const auto find_by_name = [&](const std::string& name) -> const Field* {
294
804k
        const auto value_it = partition_values.find(name);
295
804k
        return value_it == partition_values.end() ? nullptr : &value_it->second;
296
804k
    };
297
545k
    if (const auto* value = find_by_name(table_column.name); value != nullptr) {
298
28.4k
        return value;
299
28.4k
    }
300
517k
    if (table_column.has_identifier_name()) {
301
259k
        if (const auto* value = find_by_name(table_column.get_identifier_name());
302
259k
            value != nullptr) {
303
0
            return value;
304
0
        }
305
259k
    }
306
517k
    for (const auto& alias : table_column.name_mapping) {
307
24
        if (const auto* value = find_by_name(alias); value != nullptr) {
308
1
            return value;
309
1
        }
310
24
    }
311
517k
    return nullptr;
312
517k
}
313
314
struct FileSlotRewriteInfo {
315
    size_t block_position = 0;
316
    DataTypePtr file_type;
317
    DataTypePtr table_type;
318
    std::string file_column_name;
319
};
320
321
struct RewriteContext {
322
    RuntimeState* runtime_state = nullptr;
323
    std::vector<VExprSPtr> created_exprs {};
324
325
122k
    void add_created_expr(VExprSPtr expr) { created_exprs.push_back(std::move(expr)); }
326
327
70.2k
    Status prepare_created_exprs(VExprContext* context) const {
328
70.2k
        DORIS_CHECK(context != nullptr);
329
70.2k
        RowDescriptor row_desc;
330
122k
        for (const auto& expr : created_exprs) {
331
122k
            if (dynamic_cast<const Cast*>(expr.get()) != nullptr && runtime_state == nullptr) {
332
0
                return Status::InvalidArgument(
333
0
                        "RuntimeState is required to prepare rewritten cast expression {}",
334
0
                        expr->expr_name());
335
0
            }
336
122k
            RETURN_IF_ERROR(expr->prepare(runtime_state, row_desc, context));
337
122k
        }
338
70.2k
        return Status::OK();
339
70.2k
    }
340
};
341
342
static VExprSPtr create_file_slot_ref(const VSlotRef& slot_ref,
343
                                      const FileSlotRewriteInfo& rewrite_info,
344
71.9k
                                      RewriteContext* rewrite_context) {
345
71.9k
    auto ref =
346
71.9k
            VSlotRef::create_shared(slot_ref.slot_id(), cast_set<int>(rewrite_info.block_position),
347
71.9k
                                    -1, rewrite_info.file_type, rewrite_info.file_column_name);
348
71.9k
    rewrite_context->add_created_expr(ref);
349
71.9k
    return ref;
350
71.9k
}
351
352
170k
static bool is_cast_expr(const VExprSPtr& expr) {
353
170k
    return dynamic_cast<const Cast*>(expr.get()) != nullptr;
354
170k
}
355
356
193k
static bool is_binary_comparison_predicate(const VExprSPtr& expr) {
357
193k
    if (expr == nullptr || expr->get_num_children() != 2 ||
358
193k
        (expr->node_type() != TExprNodeType::BINARY_PRED &&
359
132k
         expr->node_type() != TExprNodeType::NULL_AWARE_BINARY_PRED)) {
360
132k
        return false;
361
132k
    }
362
61.4k
    switch (expr->op()) {
363
13.9k
    case TExprOpcode::EQ:
364
13.9k
    case TExprOpcode::EQ_FOR_NULL:
365
15.7k
    case TExprOpcode::NE:
366
18.6k
    case TExprOpcode::GE:
367
57.4k
    case TExprOpcode::GT:
368
59.5k
    case TExprOpcode::LE:
369
61.4k
    case TExprOpcode::LT:
370
61.4k
        return true;
371
0
    default:
372
0
        return false;
373
61.4k
    }
374
61.4k
}
375
376
16
std::string TableColumnMapperOptions::debug_string() const {
377
16
    std::ostringstream out;
378
16
    out << "TableColumnMapperOptions{mode=" << mapping_mode_to_string(mode)
379
16
        << ", reject_missing_required_field=" << reject_missing_required_field
380
16
        << ", allow_idless_complex_wrapper_projection=" << allow_idless_complex_wrapper_projection
381
16
        << ", enable_row_lineage_virtual_columns=" << enable_row_lineage_virtual_columns << "}";
382
16
    return out.str();
383
16
}
384
385
11
std::string ColumnDefinition::debug_string() const {
386
11
    std::ostringstream out;
387
11
    out << "ColumnDefinition{name=" << name << ", identifier=" << field_debug_string(identifier)
388
11
        << ", name_mapping="
389
11
        << join_debug_strings(name_mapping, [](const std::string& name) { return name; })
390
11
        << ", has_name_mapping=" << has_name_mapping << ", local_id=" << local_id
391
11
        << ", timestamp_is_adjusted_to_utc="
392
11
        << (timestamp_is_adjusted_to_utc.has_value()
393
11
                    ? (*timestamp_is_adjusted_to_utc ? "true" : "false")
394
11
                    : "unset")
395
11
        << ", type=" << data_type_debug_string(type) << ", children="
396
11
        << join_debug_strings(children,
397
11
                              [](const ColumnDefinition& child) { return child.debug_string(); })
398
11
        << ", identity_children="
399
11
        << join_debug_strings(identity_children,
400
11
                              [](const ColumnDefinition& child) { return child.debug_string(); })
401
11
        << ", has_default_expr=" << (default_expr != nullptr)
402
11
        << ", has_initial_default=" << initial_default_value.has_value() << ", is_optional=";
403
11
    if (is_optional.has_value()) {
404
0
        out << *is_optional;
405
11
    } else {
406
11
        out << "unknown";
407
11
    }
408
11
    out << ", is_partition_key=" << is_partition_key << "}";
409
11
    return out.str();
410
11
}
411
412
2
std::string LocalColumnIndex::debug_string() const {
413
2
    std::ostringstream out;
414
2
    out << "LocalColumnIndex{index=" << index << ", project_all_children=" << project_all_children
415
2
        << ", timestamp_is_adjusted_to_utc="
416
2
        << (timestamp_is_adjusted_to_utc.has_value()
417
2
                    ? (*timestamp_is_adjusted_to_utc ? "true" : "false")
418
2
                    : "unset")
419
2
        << ", children="
420
2
        << join_debug_strings(children,
421
2
                              [](const LocalColumnIndex& child) { return child.debug_string(); })
422
2
        << "}";
423
2
    return out.str();
424
2
}
425
426
12
std::string ColumnMapping::debug_string() const {
427
12
    std::ostringstream out;
428
12
    out << "ColumnMapping{global_index=" << global_index
429
12
        << ", table_column_name=" << table_column_name << ", file_local_id=";
430
12
    if (file_local_id.has_value()) {
431
7
        out << *file_local_id;
432
7
    } else {
433
5
        out << "null";
434
5
    }
435
12
    out << ", constant_index=";
436
12
    if (constant_index.has_value()) {
437
5
        out << *constant_index;
438
7
    } else {
439
7
        out << "null";
440
7
    }
441
12
    out << ", file_column_name=" << file_column_name << ", timestamp_is_adjusted_to_utc="
442
12
        << (timestamp_is_adjusted_to_utc.has_value()
443
12
                    ? (*timestamp_is_adjusted_to_utc ? "true" : "false")
444
12
                    : "unset")
445
12
        << ", original_file_type=" << data_type_debug_string(original_file_type)
446
12
        << ", original_file_children="
447
12
        << join_debug_strings(original_file_children,
448
12
                              [](const ColumnDefinition& child) { return child.debug_string(); })
449
12
        << ", file_type=" << data_type_debug_string(file_type)
450
12
        << ", table_type=" << data_type_debug_string(table_type)
451
12
        << ", has_projection=" << (projection != nullptr) << ", child_mappings="
452
12
        << join_debug_strings(child_mappings,
453
12
                              [](const ColumnMapping& child) { return child.debug_string(); })
454
12
        << ", is_trivial=" << is_trivial << ", is_constant=" << constant_index.has_value()
455
12
        << ", filter_conversion=" << filter_conversion_type_to_string(filter_conversion)
456
12
        << ", virtual_column_type=" << virtual_column_type_to_string(virtual_column_type)
457
12
        << ", has_default_expr=" << (default_expr != nullptr) << "}";
458
12
    return out.str();
459
12
}
460
461
1
std::string TableColumnMapper::debug_string() const {
462
1
    std::ostringstream out;
463
1
    out << "TableColumnMapper{options=" << _options.debug_string() << ", mappings="
464
1
        << join_debug_strings(_mappings,
465
2
                              [](const ColumnMapping& mapping) { return mapping.debug_string(); })
466
1
        << ", hidden_mappings="
467
1
        << join_debug_strings(_hidden_mappings,
468
1
                              [](const ColumnMapping& mapping) { return mapping.debug_string(); })
469
1
        << ", constant_count=" << _constant_map.size() << "}";
470
1
    return out.str();
471
1
}
472
473
static const FileSlotRewriteInfo* find_slot_rewrite_info(
474
        const VExprSPtr& expr,
475
        const std::map<GlobalIndex, FileSlotRewriteInfo>& global_to_file_slot,
476
62.2k
        const VSlotRef** slot_ref) {
477
62.2k
    if (expr == nullptr) {
478
0
        return nullptr;
479
0
    }
480
62.2k
    VExprSPtr slot_expr = expr;
481
62.2k
    const bool input_is_cast = is_cast_expr(expr) && expr->get_num_children() == 1;
482
62.2k
    if (is_cast_expr(expr) && expr->get_num_children() == 1) {
483
442
        slot_expr = expr->children()[0];
484
442
    }
485
62.2k
    if (!slot_expr->is_slot_ref()) {
486
13.5k
        return nullptr;
487
13.5k
    }
488
48.6k
    const auto* candidate_slot_ref = assert_cast<const VSlotRef*>(slot_expr.get());
489
48.6k
    const auto rewrite_it = global_to_file_slot.find(slot_ref_global_index(*candidate_slot_ref));
490
48.6k
    if (rewrite_it == global_to_file_slot.end()) {
491
0
        return nullptr;
492
0
    }
493
48.6k
    if (input_is_cast && !expr->data_type()->equals(*rewrite_it->second.table_type)) {
494
426
        return nullptr;
495
426
    }
496
48.2k
    if (slot_ref != nullptr) {
497
48.2k
        *slot_ref = candidate_slot_ref;
498
48.2k
    }
499
48.2k
    return &rewrite_it->second;
500
48.6k
}
501
502
715k
static bool filter_conversion_has_local_source(FilterConversionType conversion) {
503
715k
    switch (conversion) {
504
683k
    case FilterConversionType::COPY_DIRECTLY:
505
697k
    case FilterConversionType::CAST_FILTER:
506
713k
    case FilterConversionType::READER_EXPRESSION:
507
713k
        return true;
508
2.26k
    case FilterConversionType::FINALIZE_ONLY:
509
2.26k
    case FilterConversionType::CONSTANT:
510
2.26k
        return false;
511
715k
    }
512
0
    return false;
513
715k
}
514
515
static bool table_filter_has_only_local_entries(
516
79.8k
        const TableFilter& table_filter, const std::map<GlobalIndex, FilterEntry>& filter_entries) {
517
80.8k
    for (const auto global_index : table_filter.global_indices) {
518
80.8k
        const auto entry_it = filter_entries.find(global_index);
519
80.8k
        if (entry_it == filter_entries.end() || !entry_it->second.is_local()) {
520
6.98k
            return false;
521
6.98k
        }
522
80.8k
    }
523
72.8k
    return true;
524
79.8k
}
525
526
static VExprSPtr unwrap_literal_for_file_cast(const VExprSPtr& expr,
527
53.9k
                                              const DataTypePtr& table_type) {
528
53.9k
    if (expr == nullptr) {
529
0
        return nullptr;
530
0
    }
531
53.9k
    if (expr->is_literal()) {
532
53.9k
        return expr;
533
53.9k
    }
534
18
    if (is_cast_expr(expr) && expr->get_num_children() == 1 && expr->children()[0]->is_literal() &&
535
18
        expr->children()[0]->data_type()->equals(*table_type)) {
536
0
        return expr->children()[0];
537
0
    }
538
18
    return nullptr;
539
18
}
540
541
0
static Field literal_field_from_expr(const VExpr& literal_expr) {
542
0
    DORIS_CHECK(literal_expr.is_literal());
543
0
    const auto* literal = dynamic_cast<const VLiteral*>(&literal_expr);
544
0
    DORIS_CHECK(literal != nullptr);
545
0
    Field field;
546
0
    literal->get_column_ptr()->get(0, field);
547
0
    return field;
548
0
}
549
550
// Table filter localization clones an already-prepared table expr and then rewrites it to file
551
// slots. Only split-local literals and BE cast nodes need table-reader-specific clone behavior;
552
// plain slot refs and literals use their own VExpr::clone_node().
553
841k
static Status clone_table_expr_node(const VExpr& expr, VExprSPtr* cloned_expr) {
554
841k
    DORIS_CHECK(cloned_expr != nullptr);
555
841k
    if (const auto* split_literal = dynamic_cast<const SplitLocalFileLiteral*>(&expr)) {
556
0
        *cloned_expr = std::make_shared<SplitLocalFileLiteral>(
557
0
                split_literal->data_type(), literal_field_from_expr(expr),
558
0
                split_literal->original_type(), split_literal->original_field());
559
841k
    } else if (const auto* vcast_expr = dynamic_cast<const VCastExpr*>(&expr);
560
841k
               vcast_expr != nullptr && vcast_expr->node_type() == TExprNodeType::CAST_EXPR) {
561
4.44k
        *cloned_expr = Cast::create_shared(vcast_expr->data_type());
562
4.44k
    }
563
841k
    return Status::OK();
564
841k
}
565
566
285k
Status clone_table_expr_tree(const VExprSPtr& expr, VExprSPtr* cloned_expr) {
567
285k
    DORIS_CHECK(cloned_expr != nullptr);
568
285k
    if (expr == nullptr) {
569
0
        *cloned_expr = nullptr;
570
0
        return Status::OK();
571
0
    }
572
285k
    return expr->deep_clone(cloned_expr, clone_table_expr_node);
573
285k
}
574
575
static VExprSPtr original_table_literal(const VExprSPtr& literal_expr,
576
53.9k
                                        RewriteContext* rewrite_context = nullptr) {
577
53.9k
    DORIS_CHECK(literal_expr != nullptr);
578
53.9k
    DORIS_CHECK(literal_expr->is_literal());
579
53.9k
    const auto* rewritten_literal = dynamic_cast<const SplitLocalFileLiteral*>(literal_expr.get());
580
53.9k
    if (rewritten_literal == nullptr) {
581
53.9k
        return literal_expr;
582
53.9k
    }
583
2
    auto literal = VLiteral::create_shared(rewritten_literal->original_type(),
584
2
                                           rewritten_literal->original_field());
585
2
    if (rewrite_context != nullptr) {
586
0
        rewrite_context->add_created_expr(literal);
587
0
    }
588
2
    return literal;
589
53.9k
}
590
591
81.5k
static ColumnDefinition hidden_column_from_slot_ref(const VSlotRef& slot_ref) {
592
81.5k
    ColumnDefinition column;
593
81.5k
    column.name = slot_ref.column_name();
594
81.5k
    column.identifier = Field::create_field<TYPE_STRING>(column.name);
595
81.5k
    column.type = slot_ref.data_type();
596
81.5k
    return column;
597
81.5k
}
598
599
static void collect_top_level_slot_columns(const VExprSPtr& expr,
600
253k
                                           std::map<GlobalIndex, ColumnDefinition>* columns) {
601
253k
    DORIS_CHECK(columns != nullptr);
602
253k
    if (expr == nullptr) {
603
0
        return;
604
0
    }
605
253k
    if (expr->is_slot_ref()) {
606
81.5k
        const auto* slot_ref = assert_cast<const VSlotRef*>(expr.get());
607
81.5k
        columns->try_emplace(slot_ref_global_index(*slot_ref),
608
81.5k
                             hidden_column_from_slot_ref(*slot_ref));
609
81.5k
        return;
610
81.5k
    }
611
173k
    for (const auto& child : expr->children()) {
612
173k
        collect_top_level_slot_columns(child, columns);
613
173k
    }
614
171k
}
615
616
486
static std::optional<uint8_t> signed_integer_width(PrimitiveType type) {
617
486
    switch (type) {
618
0
    case TYPE_TINYINT:
619
0
        return 8;
620
4
    case TYPE_SMALLINT:
621
4
        return 16;
622
185
    case TYPE_INT:
623
185
        return 32;
624
164
    case TYPE_BIGINT:
625
164
        return 64;
626
0
    case TYPE_LARGEINT:
627
0
        return 128;
628
133
    default:
629
133
        return std::nullopt;
630
486
    }
631
486
}
632
633
117
static std::optional<uint8_t> floating_width(PrimitiveType type) {
634
117
    switch (type) {
635
0
    case TYPE_FLOAT:
636
0
        return 32;
637
1
    case TYPE_DOUBLE:
638
1
        return 64;
639
116
    default:
640
116
        return std::nullopt;
641
117
    }
642
117
}
643
644
17
static std::optional<uint8_t> floating_exact_integer_width(PrimitiveType type) {
645
17
    switch (type) {
646
0
    case TYPE_FLOAT:
647
0
        return 24;
648
3
    case TYPE_DOUBLE:
649
3
        return 53;
650
14
    default:
651
14
        return std::nullopt;
652
17
    }
653
17
}
654
655
static bool is_lossless_file_to_table_numeric_cast(const DataTypePtr& file_type,
656
49.2k
                                                   const DataTypePtr& table_type) {
657
49.2k
    const auto file_nested_type = remove_nullable(file_type);
658
49.2k
    const auto table_nested_type = remove_nullable(table_type);
659
49.2k
    if (file_nested_type->equals(*table_nested_type)) {
660
48.9k
        return true;
661
48.9k
    }
662
663
299
    const auto file_primitive_type = file_nested_type->get_primitive_type();
664
299
    const auto table_primitive_type = table_nested_type->get_primitive_type();
665
299
    if (const auto file_width = signed_integer_width(file_primitive_type)) {
666
185
        if (const auto table_width = signed_integer_width(table_primitive_type)) {
667
168
            return *table_width >= *file_width;
668
168
        }
669
17
        if (const auto table_width = floating_exact_integer_width(table_primitive_type)) {
670
3
            return *table_width >= *file_width;
671
3
        }
672
14
        return false;
673
17
    }
674
114
    if (const auto file_width = floating_width(file_primitive_type)) {
675
1
        const auto table_width = floating_width(table_primitive_type);
676
1
        return table_width.has_value() && *table_width >= *file_width;
677
1
    }
678
113
    return false;
679
114
}
680
681
static VExprSPtr rewrite_literal_to_file_type(const VExprSPtr& literal_expr,
682
                                              const FileSlotRewriteInfo& rewrite_info,
683
52.4k
                                              RewriteContext* rewrite_context) {
684
52.4k
    DORIS_CHECK(literal_expr != nullptr);
685
52.4k
    DORIS_CHECK(literal_expr->is_literal());
686
52.4k
    const auto original_literal = original_table_literal(literal_expr, rewrite_context);
687
52.4k
    const Field original_field = literal_field(original_literal);
688
52.4k
    if (rewrite_info.file_type->equals(*original_literal->data_type())) {
689
3.19k
        return original_literal;
690
3.19k
    }
691
    // A literal round trip alone cannot prove that file-local evaluation is safe: the file slot
692
    // itself may lose information when materialized as the table type. For example, DOUBLE 1.5
693
    // becomes BIGINT 1, so table predicate `value = 1` is true while file predicate
694
    // `value = 1.0` is false. Complex Field equality also does not compare nested contents.
695
    // Restrict localization to scalar numeric casts that preserve every file value; unsupported
696
    // and complex casts keep the table predicate and evaluate after materialization.
697
49.2k
    if (!is_lossless_file_to_table_numeric_cast(rewrite_info.file_type,
698
49.2k
                                                original_literal->data_type())) {
699
153
        return nullptr;
700
153
    }
701
49.0k
    Field file_field;
702
49.0k
    try {
703
49.0k
        convert_field_to_type(original_field, *rewrite_info.file_type, &file_field,
704
49.0k
                              original_literal->data_type().get());
705
49.0k
    } catch (const Exception&) {
706
0
        return nullptr;
707
0
    }
708
49.0k
    if (file_field.is_null()) {
709
56
        return nullptr;
710
56
    }
711
49.0k
    if (file_field.get_type() != remove_nullable(rewrite_info.file_type)->get_primitive_type()) {
712
50
        return nullptr;
713
50
    }
714
48.9k
    Field round_trip_field;
715
48.9k
    try {
716
48.9k
        convert_field_to_type(file_field, *original_literal->data_type(), &round_trip_field,
717
48.9k
                              rewrite_info.file_type.get());
718
48.9k
    } catch (const Exception&) {
719
1.18k
        return nullptr;
720
1.18k
    }
721
    // The file-to-table type check protects every possible file value. This round trip separately
722
    // proves that the specific predicate boundary is exactly representable in the file type.
723
47.8k
    if (round_trip_field != original_field) {
724
2
        return nullptr;
725
2
    }
726
47.8k
    auto literal = std::make_shared<SplitLocalFileLiteral>(
727
47.8k
            rewrite_info.file_type, file_field, original_literal->data_type(), original_field);
728
47.8k
    rewrite_context->add_created_expr(literal);
729
47.8k
    return literal;
730
47.8k
}
731
732
static bool rewrite_binary_slot_literal_predicate(
733
        const VExprSPtr& expr,
734
        const std::map<GlobalIndex, FileSlotRewriteInfo>& global_to_file_slot,
735
120k
        RewriteContext* rewrite_context) {
736
120k
    if (!is_binary_comparison_predicate(expr)) {
737
67.1k
        return false;
738
67.1k
    }
739
53.1k
    auto children = expr->children();
740
53.1k
    const VSlotRef* slot_ref = nullptr;
741
53.1k
    const FileSlotRewriteInfo* rewrite_info =
742
53.1k
            find_slot_rewrite_info(children[0], global_to_file_slot, &slot_ref);
743
53.1k
    int slot_child_idx = 0;
744
53.1k
    int literal_child_idx = 1;
745
53.1k
    if (rewrite_info == nullptr) {
746
6.96k
        rewrite_info = find_slot_rewrite_info(children[1], global_to_file_slot, &slot_ref);
747
6.96k
        slot_child_idx = 1;
748
6.96k
        literal_child_idx = 0;
749
6.96k
    }
750
53.1k
    if (rewrite_info == nullptr || slot_ref == nullptr) {
751
6.96k
        return false;
752
6.96k
    }
753
46.1k
    auto literal_expr =
754
46.1k
            unwrap_literal_for_file_cast(children[literal_child_idx], rewrite_info->table_type);
755
46.1k
    if (literal_expr == nullptr) {
756
18
        return false;
757
18
    }
758
759
46.1k
    auto rewritten_literal =
760
46.1k
            rewrite_literal_to_file_type(literal_expr, *rewrite_info, rewrite_context);
761
46.1k
    if (rewritten_literal == nullptr) {
762
1.36k
        children[literal_child_idx] = original_table_literal(literal_expr, rewrite_context);
763
1.36k
        expr->set_children(std::move(children));
764
1.36k
        return false;
765
1.36k
    }
766
767
44.7k
    children[slot_child_idx] = create_file_slot_ref(*slot_ref, *rewrite_info, rewrite_context);
768
44.7k
    children[literal_child_idx] = std::move(rewritten_literal);
769
44.7k
    expr->set_children(std::move(children));
770
44.7k
    return true;
771
46.1k
}
772
773
static bool rewrite_in_slot_literal_predicate(
774
        const VExprSPtr& expr,
775
        const std::map<GlobalIndex, FileSlotRewriteInfo>& global_to_file_slot,
776
75.5k
        RewriteContext* rewrite_context) {
777
75.5k
    if (expr->node_type() != TExprNodeType::IN_PRED || expr->get_num_children() < 2) {
778
73.2k
        return false;
779
73.2k
    }
780
2.22k
    auto children = expr->children();
781
2.22k
    const VSlotRef* slot_ref = nullptr;
782
2.22k
    const FileSlotRewriteInfo* rewrite_info =
783
2.22k
            find_slot_rewrite_info(children[0], global_to_file_slot, &slot_ref);
784
2.22k
    if (rewrite_info == nullptr || slot_ref == nullptr) {
785
76
        return false;
786
76
    }
787
788
2.14k
    VExprSPtrs rewritten_literals;
789
2.14k
    rewritten_literals.reserve(children.size() - 1);
790
7.63k
    for (size_t child_idx = 1; child_idx < children.size(); ++child_idx) {
791
5.53k
        auto literal_expr =
792
5.53k
                unwrap_literal_for_file_cast(children[child_idx], rewrite_info->table_type);
793
5.53k
        if (literal_expr == nullptr) {
794
0
            return false;
795
0
        }
796
5.53k
        auto rewritten_literal =
797
5.53k
                rewrite_literal_to_file_type(literal_expr, *rewrite_info, rewrite_context);
798
5.53k
        if (rewritten_literal == nullptr) {
799
150
            for (size_t restore_idx = 1; restore_idx < children.size(); ++restore_idx) {
800
100
                auto restore_literal = unwrap_literal_for_file_cast(children[restore_idx],
801
100
                                                                    rewrite_info->table_type);
802
100
                if (restore_literal != nullptr) {
803
100
                    children[restore_idx] =
804
100
                            original_table_literal(restore_literal, rewrite_context);
805
100
                }
806
100
            }
807
50
            expr->set_children(std::move(children));
808
50
            return false;
809
50
        }
810
5.48k
        rewritten_literals.push_back(std::move(rewritten_literal));
811
5.48k
    }
812
813
2.09k
    children[0] = create_file_slot_ref(*slot_ref, *rewrite_info, rewrite_context);
814
7.58k
    for (size_t literal_idx = 0; literal_idx < rewritten_literals.size(); ++literal_idx) {
815
5.48k
        children[literal_idx + 1] = std::move(rewritten_literals[literal_idx]);
816
5.48k
    }
817
2.09k
    expr->set_children(std::move(children));
818
2.09k
    return true;
819
2.14k
}
820
821
static VExprSPtr create_file_struct_child_name_literal(const std::string& file_child_name,
822
1.41k
                                                       RewriteContext* rewrite_context) {
823
1.41k
    auto literal = VLiteral::create_shared(std::make_shared<DataTypeString>(),
824
1.41k
                                           Field::create_field<TYPE_STRING>(file_child_name));
825
1.41k
    rewrite_context->add_created_expr(literal);
826
1.41k
    return literal;
827
1.41k
}
828
829
static bool needs_complex_file_slot_cast(const DataTypePtr& file_type,
830
2.05k
                                         const DataTypePtr& table_type) {
831
2.05k
    if (file_type == nullptr || table_type == nullptr || file_type->equals(*table_type)) {
832
0
        return false;
833
0
    }
834
2.05k
    const auto file_nested_type = remove_nullable(file_type);
835
2.05k
    const auto table_nested_type = remove_nullable(table_type);
836
2.05k
    if (file_nested_type->equals(*table_nested_type)) {
837
0
        return false;
838
0
    }
839
2.05k
    return is_complex_type(file_nested_type->get_primitive_type()) ||
840
2.05k
           is_complex_type(table_nested_type->get_primitive_type());
841
2.05k
}
842
843
1.41k
static bool collect_struct_element_chain(const VExprSPtr& expr, std::vector<VExprSPtr>* chain) {
844
1.41k
    DORIS_CHECK(chain != nullptr);
845
1.41k
    if (!is_struct_element_expr(expr)) {
846
0
        return false;
847
0
    }
848
1.41k
    const auto& parent = expr->children()[0];
849
1.41k
    if (is_struct_element_expr(parent)) {
850
157
        if (!collect_struct_element_chain(parent, chain)) {
851
0
            return false;
852
0
        }
853
1.26k
    } else if (!parent->is_slot_ref()) {
854
        // Only support file-local rewrite for struct child chains rooted directly at a top-level
855
        // slot, for example `element_at(s, 'a')` or `element_at(element_at(s, 'a'), 'b')`.
856
        //
857
        // Do not localize computed complex parents such as
858
        // `element_at(element_at(map_values(m), 1), 'full_name')`. The intermediate map/array
859
        // result has already been reshaped by scan projection and may have a different child order
860
        // from the table expression. Partially rewriting that expression against the file block can
861
        // silently evaluate the wrong struct child and filter out valid rows. Those predicates must
862
        // remain as table-level conjuncts and be evaluated after TableReader materialization.
863
0
        return false;
864
0
    }
865
1.41k
    chain->push_back(expr);
866
1.41k
    return true;
867
1.41k
}
868
869
static bool can_filter_before_table_nullability_alignment(const DataTypePtr& file_type,
870
2.67k
                                                          const DataTypePtr& table_type) {
871
2.67k
    DORIS_CHECK(file_type != nullptr);
872
2.67k
    DORIS_CHECK(table_type != nullptr);
873
    // File-local conjuncts run before TableReader validates the materialized table schema. A
874
    // nullable file value mapped to a required table value must therefore reach
875
    // _align_column_nullability(). For example, with file STRUCT<a: Nullable(INT)>, table
876
    // STRUCT<a: BIGINT>, rows [NULL, 20], and `s.a > 10`, filtering in the file domain would drop
877
    // NULL first and hide the table-contract violation. The reverse direction is safe: a required
878
    // file value can always be wrapped as a nullable table value after filtering.
879
2.67k
    return !file_type->is_nullable() || table_type->is_nullable();
880
2.67k
}
881
882
static bool rewrite_struct_element_path_to_file_expr(
883
        const VExprSPtr& expr, const std::vector<ColumnMapping>& mappings,
884
        const std::map<GlobalIndex, FileSlotRewriteInfo>& global_to_file_slot,
885
4.65k
        RewriteContext* rewrite_context) {
886
4.65k
    ResolvedNestedStructPath resolved;
887
4.65k
    if (!resolve_nested_struct_expr_for_file(expr, mappings, &resolved)) {
888
3.39k
        return false;
889
3.39k
    }
890
891
1.26k
    std::vector<VExprSPtr> struct_element_chain;
892
1.26k
    if (!collect_struct_element_chain(expr, &struct_element_chain) ||
893
1.26k
        struct_element_chain.size() != resolved.file_child_names.size() ||
894
1.26k
        struct_element_chain.size() != resolved.file_child_types.size()) {
895
0
        return false;
896
0
    }
897
898
1.26k
    auto root_children = struct_element_chain.front()->children();
899
1.26k
    if (!root_children[0]->is_slot_ref()) {
900
0
        return false;
901
0
    }
902
1.26k
    const auto* slot_ref = assert_cast<const VSlotRef*>(root_children[0].get());
903
1.26k
    const auto rewrite_it = global_to_file_slot.find(slot_ref_global_index(*slot_ref));
904
1.26k
    if (rewrite_it == global_to_file_slot.end()) {
905
0
        return false;
906
0
    }
907
908
    // Check every value-producing level, including the root struct. A nullable parent also makes
909
    // a child access nullable even when the child type itself is required, so checking only the
910
    // final leaf is insufficient. If any file level is more nullable than its table counterpart,
911
    // keep the complete predicate above TableReader so schema validation observes all NULLs before
912
    // row filtering.
913
1.26k
    if (!can_filter_before_table_nullability_alignment(rewrite_it->second.file_type,
914
1.26k
                                                       rewrite_it->second.table_type)) {
915
0
        return false;
916
0
    }
917
2.67k
    for (size_t idx = 0; idx < struct_element_chain.size(); ++idx) {
918
1.41k
        if (!can_filter_before_table_nullability_alignment(
919
1.41k
                    resolved.file_child_types[idx], struct_element_chain[idx]->data_type())) {
920
2
            return false;
921
2
        }
922
1.41k
    }
923
924
    // File-local conjuncts are prepared against the file-reader Block, so both the root slot and
925
    // every struct selector must be expressed in file schema terms. For a renamed Iceberg field,
926
    // keeping the table selector would prepare `element_at(file_struct<rename>, 'renamed')` and
927
    // fail before any rows are read. Rewrite the whole chain while ColumnMapping still preserves
928
    // the table-to-file relationship. Example:
929
    //   table filter: element_at(element_at(s, 'renamed_parent'), 'renamed_leaf')
930
    //   old file:     s<parent<leaf>>
931
    //   file filter:  element_at(element_at(s, 'parent'), 'leaf')
932
1.25k
    root_children[0] = create_file_slot_ref(*slot_ref, rewrite_it->second, rewrite_context);
933
1.25k
    struct_element_chain.front()->set_children(std::move(root_children));
934
2.67k
    for (size_t idx = 0; idx < struct_element_chain.size(); ++idx) {
935
1.41k
        auto children = struct_element_chain[idx]->children();
936
1.41k
        children[1] = create_file_struct_child_name_literal(resolved.file_child_names[idx],
937
1.41k
                                                            rewrite_context);
938
1.41k
        struct_element_chain[idx]->set_children(std::move(children));
939
        // The selector name and the expression return type must be moved to file schema together.
940
        // Example:
941
        //   table filter: element_at(element_at(s, 'new_a'), 'new_aa') = 50
942
        //   old file:     s.new_a STRUCT<aa, bb>
943
        //   file filter:  element_at(element_at(s, 'new_a'), 'aa') = 50
944
        //
945
        // If the inner element_at keeps the table return type STRUCT<new_aa, bb>, preparing the
946
        // outer element_at(..., 'aa') fails before scanning because `aa` is not a table field.
947
1.41k
        struct_element_chain[idx]->data_type() = resolved.file_child_types[idx];
948
1.41k
    }
949
1.25k
    return true;
950
1.26k
}
951
952
static VExprSPtr cast_file_expr_to_table_type(const VExprSPtr& file_expr,
953
                                              const DataTypePtr& table_type,
954
1.49k
                                              RewriteContext* rewrite_context) {
955
1.49k
    DORIS_CHECK(file_expr != nullptr);
956
1.49k
    DORIS_CHECK(table_type != nullptr);
957
1.49k
    DORIS_CHECK(rewrite_context != nullptr);
958
1.49k
    auto cast_expr = Cast::create_shared(table_type);
959
1.49k
    cast_expr->add_child(file_expr);
960
1.49k
    rewrite_context->add_created_expr(cast_expr);
961
1.49k
    return cast_expr;
962
1.49k
}
963
964
// Prefer comparing in the physical file leaf type when a table predicate uses a promoted struct
965
// child. For example, with table STRUCT<a: BIGINT>, old-file STRUCT<a: INT>, and `s.a = 10`, the
966
// localized predicate should be `file_s.a::INT = 10::INT`, not
967
// `CAST(file_s.a::INT AS BIGINT) = 10::BIGINT`. Converting one literal avoids a cast for every row.
968
//
969
// This rewrite is valid only when every possible file value survives file-to-table conversion and
970
// the particular literal survives a table-to-file-to-table round trip. A value such as BIGINT
971
// 2147483648 cannot be represented by an INT file leaf, so that case deliberately falls back to
972
// `CAST(file_s.a AS BIGINT) = 2147483648`, which preserves the original table-level semantics.
973
static bool rewrite_binary_struct_literal_predicate(
974
        const VExprSPtr& expr, const std::vector<ColumnMapping>& filter_mappings,
975
        const std::map<GlobalIndex, FileSlotRewriteInfo>& global_to_file_slot,
976
73.4k
        RewriteContext* rewrite_context, bool* can_localize) {
977
73.4k
    DORIS_CHECK(can_localize != nullptr);
978
73.4k
    if (!is_binary_comparison_predicate(expr)) {
979
65.0k
        return false;
980
65.0k
    }
981
8.36k
    auto children = expr->children();
982
8.36k
    int struct_child_idx = -1;
983
8.36k
    int literal_child_idx = -1;
984
8.36k
    if (is_struct_element_expr(children[0])) {
985
2.19k
        struct_child_idx = 0;
986
2.19k
        literal_child_idx = 1;
987
6.16k
    } else if (is_struct_element_expr(children[1])) {
988
2
        struct_child_idx = 1;
989
2
        literal_child_idx = 0;
990
6.16k
    } else {
991
6.16k
        return false;
992
6.16k
    }
993
994
2.19k
    const auto table_leaf_type = children[struct_child_idx]->data_type();
995
2.19k
    DORIS_CHECK(table_leaf_type != nullptr);
996
2.19k
    auto table_literal = unwrap_literal_for_file_cast(children[literal_child_idx], table_leaf_type);
997
2.19k
    if (table_literal == nullptr ||
998
2.19k
        !rewrite_struct_element_path_to_file_expr(children[struct_child_idx], filter_mappings,
999
2.19k
                                                  global_to_file_slot, rewrite_context)) {
1000
1.39k
        return false;
1001
1.39k
    }
1002
1003
802
    const auto file_leaf_type = children[struct_child_idx]->data_type();
1004
802
    DORIS_CHECK(file_leaf_type != nullptr);
1005
802
    const FileSlotRewriteInfo leaf_rewrite_info {
1006
802
            .block_position = 0,
1007
802
            .file_type = file_leaf_type,
1008
802
            .table_type = table_leaf_type,
1009
802
            .file_column_name = {},
1010
802
    };
1011
802
    auto file_literal =
1012
802
            rewrite_literal_to_file_type(table_literal, leaf_rewrite_info, rewrite_context);
1013
802
    if (file_literal != nullptr) {
1014
768
        children[literal_child_idx] = std::move(file_literal);
1015
768
    } else {
1016
34
        if (!is_lossless_file_to_table_numeric_cast(file_leaf_type, table_leaf_type)) {
1017
            // A narrowing or otherwise lossy cast can fail or produce NULL while TableReader
1018
            // materializes the table schema. Evaluating it here could filter the offending row
1019
            // before that validation, so keep the complete predicate above TableReader.
1020
1
            *can_localize = false;
1021
1
            return true;
1022
1
        }
1023
33
        children[struct_child_idx] = cast_file_expr_to_table_type(children[struct_child_idx],
1024
33
                                                                  table_leaf_type, rewrite_context);
1025
33
        children[literal_child_idx] = original_table_literal(table_literal, rewrite_context);
1026
33
    }
1027
801
    expr->set_children(std::move(children));
1028
801
    return true;
1029
802
}
1030
1031
// IN must use one comparison type for its probe and every candidate. Rewrite the complete literal
1032
// set only when all values are exactly representable in the file leaf type; one unsafe value makes
1033
// the whole predicate fall back to a table-type cast. For example, an INT file leaf can evaluate
1034
// `BIGINT IN (10, 20)` as `INT IN (10, 20)`, but `BIGINT IN (10, 2147483648)` must stay BIGINT.
1035
static bool rewrite_in_struct_literal_predicate(
1036
        const VExprSPtr& expr, const std::vector<ColumnMapping>& filter_mappings,
1037
        const std::map<GlobalIndex, FileSlotRewriteInfo>& global_to_file_slot,
1038
72.5k
        RewriteContext* rewrite_context, bool* can_localize) {
1039
72.5k
    DORIS_CHECK(can_localize != nullptr);
1040
72.5k
    if (expr->node_type() != TExprNodeType::IN_PRED || expr->get_num_children() < 2 ||
1041
72.5k
        !is_struct_element_expr(expr->children()[0])) {
1042
72.5k
        return false;
1043
72.5k
    }
1044
30
    auto children = expr->children();
1045
30
    const auto table_leaf_type = children[0]->data_type();
1046
30
    DORIS_CHECK(table_leaf_type != nullptr);
1047
30
    VExprSPtrs table_literals;
1048
30
    table_literals.reserve(children.size() - 1);
1049
38
    for (size_t child_idx = 1; child_idx < children.size(); ++child_idx) {
1050
8
        auto table_literal = unwrap_literal_for_file_cast(children[child_idx], table_leaf_type);
1051
8
        if (table_literal == nullptr) {
1052
0
            return false;
1053
0
        }
1054
8
        table_literals.push_back(std::move(table_literal));
1055
8
    }
1056
30
    if (!rewrite_struct_element_path_to_file_expr(children[0], filter_mappings, global_to_file_slot,
1057
30
                                                  rewrite_context)) {
1058
0
        return false;
1059
0
    }
1060
1061
30
    const auto file_leaf_type = children[0]->data_type();
1062
30
    DORIS_CHECK(file_leaf_type != nullptr);
1063
30
    const FileSlotRewriteInfo leaf_rewrite_info {
1064
30
            .block_position = 0,
1065
30
            .file_type = file_leaf_type,
1066
30
            .table_type = table_leaf_type,
1067
30
            .file_column_name = {},
1068
30
    };
1069
30
    VExprSPtrs file_literals;
1070
30
    file_literals.reserve(table_literals.size());
1071
30
    for (const auto& table_literal : table_literals) {
1072
8
        auto file_literal =
1073
8
                rewrite_literal_to_file_type(table_literal, leaf_rewrite_info, rewrite_context);
1074
8
        if (file_literal == nullptr) {
1075
1
            if (!is_lossless_file_to_table_numeric_cast(file_leaf_type, table_leaf_type)) {
1076
0
                *can_localize = false;
1077
0
                return true;
1078
0
            }
1079
1
            children[0] =
1080
1
                    cast_file_expr_to_table_type(children[0], table_leaf_type, rewrite_context);
1081
3
            for (size_t literal_idx = 0; literal_idx < table_literals.size(); ++literal_idx) {
1082
2
                children[literal_idx + 1] =
1083
2
                        original_table_literal(table_literals[literal_idx], rewrite_context);
1084
2
            }
1085
1
            expr->set_children(std::move(children));
1086
1
            return true;
1087
1
        }
1088
7
        file_literals.push_back(std::move(file_literal));
1089
7
    }
1090
1091
35
    for (size_t literal_idx = 0; literal_idx < file_literals.size(); ++literal_idx) {
1092
6
        children[literal_idx + 1] = std::move(file_literals[literal_idx]);
1093
6
    }
1094
29
    expr->set_children(std::move(children));
1095
29
    return true;
1096
30
}
1097
1098
static VExprSPtr rewrite_struct_or_slot_expr_to_file_expr(
1099
        const VExprSPtr& expr,
1100
        const std::map<GlobalIndex, FileSlotRewriteInfo>& global_to_file_slot,
1101
        const std::vector<ColumnMapping>& filter_mappings, RewriteContext* rewrite_context,
1102
26.4k
        bool* can_localize) {
1103
26.4k
    if (is_struct_element_expr(expr)) {
1104
2.45k
        const auto table_leaf_type = expr->data_type();
1105
2.45k
        if (!rewrite_struct_element_path_to_file_expr(expr, filter_mappings, global_to_file_slot,
1106
2.45k
                                                      rewrite_context)) {
1107
            // The scanner still evaluates the original table-level conjunct after TableReader
1108
            // finalizes the output block. Skipping an unlocalizable file conjunct is therefore
1109
            // safer than preparing a partially rewritten expression against the wrong struct
1110
            // layout. In particular, do not generate file-local conjuncts for computed complex
1111
            // parents such as `element_at(element_at(map_values(m), 1), 'field')`; only direct
1112
            // slot-rooted struct chains are supported here.
1113
2.00k
            *can_localize = false;
1114
2.00k
            return expr;
1115
2.00k
        }
1116
452
        DORIS_CHECK(table_leaf_type != nullptr);
1117
452
        DORIS_CHECK(expr->data_type() != nullptr);
1118
452
        if (!expr->data_type()->equals(*table_leaf_type)) {
1119
2
            if (!is_lossless_file_to_table_numeric_cast(expr->data_type(), table_leaf_type)) {
1120
2
                *can_localize = false;
1121
2
                return expr;
1122
2
            }
1123
            // Path localization changes the leaf to the physical file type. For example, after an
1124
            // Iceberg evolution from STRUCT<a: INT> to STRUCT<a: BIGINT>, the localized old-file
1125
            // predicate is initially `element_at(file_col, 'a')::INT = 10::BIGINT`. Cast only the
1126
            // leaf back to BIGINT so the comparison has matching operands without forcing a cast
1127
            // of the entire evolved struct (whose children may also have been added or reordered).
1128
0
            return cast_file_expr_to_table_type(expr, table_leaf_type, rewrite_context);
1129
2
        }
1130
450
        return expr;
1131
452
    }
1132
1133
23.9k
    DORIS_CHECK(expr->is_slot_ref());
1134
23.9k
    const auto* slot_ref = assert_cast<const VSlotRef*>(expr.get());
1135
23.9k
    const auto rewrite_it = global_to_file_slot.find(slot_ref_global_index(*slot_ref));
1136
23.9k
    if (rewrite_it == global_to_file_slot.end()) {
1137
0
        return expr;
1138
0
    }
1139
23.9k
    const auto& rewrite_info = rewrite_it->second;
1140
23.9k
    auto file_slot = create_file_slot_ref(*slot_ref, rewrite_info, rewrite_context);
1141
23.9k
    if (rewrite_info.file_type->equals(*rewrite_info.table_type)) {
1142
21.8k
        return file_slot;
1143
21.8k
    }
1144
2.06k
    if (needs_complex_file_slot_cast(rewrite_info.file_type, rewrite_info.table_type)) {
1145
        // Generic file-local expressions cannot safely cast an evolved complex file slot back to
1146
        // the table type. For example, ARRAY_CONTAINS(MAP_KEYS(m), 'person5') only reads map keys,
1147
        // but CAST(file_m AS table_m) first forces an incompatible old value struct into the new
1148
        // layout. Keep such predicates at table level, after TableReader materializes evolution.
1149
590
        *can_localize = false;
1150
590
        return expr;
1151
590
    }
1152
1.47k
    return cast_file_expr_to_table_type(file_slot, rewrite_info.table_type, rewrite_context);
1153
2.06k
}
1154
1155
static VExprSPtr rewrite_table_expr_to_file_expr(
1156
        const VExprSPtr& expr,
1157
        const std::map<GlobalIndex, FileSlotRewriteInfo>& global_to_file_slot,
1158
        const std::vector<ColumnMapping>& filter_mappings, RewriteContext* rewrite_context,
1159
125k
        bool* can_localize) {
1160
125k
    if (expr == nullptr) {
1161
0
        return nullptr;
1162
0
    }
1163
125k
    DORIS_CHECK(rewrite_context != nullptr);
1164
125k
    DORIS_CHECK(can_localize != nullptr);
1165
125k
    if (auto* runtime_filter = dynamic_cast<RuntimeFilterExpr*>(expr.get());
1166
125k
        runtime_filter != nullptr) {
1167
4.79k
        auto impl = runtime_filter->get_impl();
1168
4.79k
        if (impl == nullptr) {
1169
0
            *can_localize = false;
1170
0
            return expr;
1171
0
        }
1172
4.79k
        auto localized_impl = rewrite_table_expr_to_file_expr(
1173
4.79k
                impl, global_to_file_slot, filter_mappings, rewrite_context, can_localize);
1174
4.79k
        if (!*can_localize) {
1175
0
            return expr;
1176
0
        }
1177
4.79k
        runtime_filter->set_impl(std::move(localized_impl));
1178
4.79k
        return expr;
1179
4.79k
    }
1180
120k
    if (rewrite_binary_slot_literal_predicate(expr, global_to_file_slot, rewrite_context)) {
1181
44.7k
        return expr;
1182
44.7k
    }
1183
75.5k
    if (rewrite_in_slot_literal_predicate(expr, global_to_file_slot, rewrite_context)) {
1184
2.06k
        return expr;
1185
2.06k
    }
1186
73.4k
    if (rewrite_binary_struct_literal_predicate(expr, filter_mappings, global_to_file_slot,
1187
73.4k
                                                rewrite_context, can_localize)) {
1188
800
        return expr;
1189
800
    }
1190
72.6k
    if (rewrite_in_struct_literal_predicate(expr, filter_mappings, global_to_file_slot,
1191
72.6k
                                            rewrite_context, can_localize)) {
1192
4
        return expr;
1193
4
    }
1194
72.6k
    if (is_struct_element_expr(expr) || expr->is_slot_ref()) {
1195
26.3k
        return rewrite_struct_or_slot_expr_to_file_expr(expr, global_to_file_slot, filter_mappings,
1196
26.3k
                                                        rewrite_context, can_localize);
1197
26.3k
    }
1198
    // The input is a split-local cloned tree. A previous split-local clone may already have
1199
    // inserted Cast(slot). Keep that rewrite idempotent: rewrite the cast child from table slot to
1200
    // the current split's file slot, and drop the cast when the current split no longer needs it.
1201
46.3k
    if (is_cast_expr(expr) && expr->get_num_children() == 1) {
1202
1.62k
        const auto& child = expr->children()[0];
1203
1.62k
        if (child->is_slot_ref()) {
1204
1.58k
            const auto* slot_ref = assert_cast<const VSlotRef*>(child.get());
1205
1.58k
            const auto rewrite_it = global_to_file_slot.find(slot_ref_global_index(*slot_ref));
1206
1.58k
            if (rewrite_it != global_to_file_slot.end() &&
1207
1.58k
                expr->data_type()->equals(*rewrite_it->second.table_type)) {
1208
1
                auto rewritten_child =
1209
1
                        create_file_slot_ref(*slot_ref, rewrite_it->second, rewrite_context);
1210
1
                if (rewrite_it->second.file_type->equals(*rewrite_it->second.table_type)) {
1211
0
                    return rewritten_child;
1212
0
                }
1213
1
                if (needs_complex_file_slot_cast(rewrite_it->second.file_type,
1214
1
                                                 rewrite_it->second.table_type)) {
1215
0
                    *can_localize = false;
1216
0
                    return expr;
1217
0
                }
1218
1
                expr->set_children({std::move(rewritten_child)});
1219
1
                return expr;
1220
1
            }
1221
1.58k
        }
1222
1.62k
    }
1223
1224
46.3k
    VExprSPtrs rewritten_children;
1225
46.3k
    rewritten_children.reserve(expr->children().size());
1226
47.3k
    for (const auto& child : expr->children()) {
1227
47.3k
        rewritten_children.push_back(rewrite_table_expr_to_file_expr(
1228
47.3k
                child, global_to_file_slot, filter_mappings, rewrite_context, can_localize));
1229
47.3k
    }
1230
46.3k
    expr->set_children(std::move(rewritten_children));
1231
46.3k
    return expr;
1232
46.3k
}
1233
1234
static constexpr const char* ROW_LINEAGE_ROW_ID = "_row_id";
1235
static constexpr const char* ROW_LINEAGE_LAST_UPDATED_SEQ_NUMBER = "_last_updated_sequence_number";
1236
static constexpr int32_t ROW_LINEAGE_ROW_ID_FIELD_ID = 2147483540;
1237
static constexpr int32_t ROW_LINEAGE_LAST_UPDATED_SEQ_NUMBER_FIELD_ID = 2147483539;
1238
1239
21
static TableVirtualColumnType row_lineage_virtual_column_type(const std::string& column_name) {
1240
21
    if (column_name == ROW_LINEAGE_ROW_ID) {
1241
2
        return TableVirtualColumnType::ROW_ID;
1242
2
    }
1243
19
    if (column_name == ROW_LINEAGE_LAST_UPDATED_SEQ_NUMBER) {
1244
2
        return TableVirtualColumnType::LAST_UPDATED_SEQUENCE_NUMBER;
1245
2
    }
1246
17
    return TableVirtualColumnType::INVALID;
1247
19
}
1248
1249
static TableVirtualColumnType row_lineage_virtual_column_type_by_field_id(
1250
111k
        const ColumnDefinition& column) {
1251
111k
    if (!column.has_identifier_field_id()) {
1252
1.09k
        return TableVirtualColumnType::INVALID;
1253
1.09k
    }
1254
110k
    switch (column.get_identifier_field_id()) {
1255
2.09k
    case ROW_LINEAGE_ROW_ID_FIELD_ID:
1256
2.09k
        return TableVirtualColumnType::ROW_ID;
1257
1.76k
    case ROW_LINEAGE_LAST_UPDATED_SEQ_NUMBER_FIELD_ID:
1258
1.76k
        return TableVirtualColumnType::LAST_UPDATED_SEQUENCE_NUMBER;
1259
106k
    default:
1260
106k
        return TableVirtualColumnType::INVALID;
1261
110k
    }
1262
110k
}
1263
1264
static TableVirtualColumnType row_lineage_virtual_column_type(const ColumnDefinition& column,
1265
111k
                                                              TableColumnMappingMode mode) {
1266
111k
    switch (mode) {
1267
111k
    case TableColumnMappingMode::BY_FIELD_ID:
1268
111k
        return row_lineage_virtual_column_type_by_field_id(column);
1269
21
    case TableColumnMappingMode::BY_NAME:
1270
21
    case TableColumnMappingMode::BY_INDEX:
1271
21
        return row_lineage_virtual_column_type(column.name);
1272
111k
    }
1273
0
    return TableVirtualColumnType::INVALID;
1274
111k
}
1275
1276
// Returns true when the current file type is not the exact nested type the scan should expose.
1277
// This is about building the projected file-side type/projection, not about whether TableReader
1278
// later needs to rematerialize the complex value back to table layout.
1279
553k
static bool needs_projected_file_type_rebuild(const ColumnMapping& mapping) {
1280
553k
    if (!is_complex_type(mapping.file_type->get_primitive_type())) {
1281
311k
        return false;
1282
311k
    }
1283
241k
    if (mapping.child_mappings.empty()) {
1284
0
        return false;
1285
0
    }
1286
241k
    DORIS_CHECK(mapping.file_type != nullptr);
1287
241k
    DORIS_CHECK(mapping.table_type != nullptr);
1288
241k
    if (remove_nullable(mapping.file_type)->get_primitive_type() !=
1289
241k
        remove_nullable(mapping.table_type)->get_primitive_type()) {
1290
0
        return true;
1291
0
    }
1292
241k
    if (!mapping.table_type->equals(*mapping.file_type)) {
1293
19.3k
        return true;
1294
19.3k
    }
1295
344k
    for (const auto& child_mapping : mapping.child_mappings) {
1296
        // Rename-only child mappings do not change the file-side projected shape. If field-id
1297
        // matching maps table child `renamed_b` to file child `b`, the file reader can still expose
1298
        // the original file type as long as child count/order/types are unchanged.
1299
344k
        if (!child_mapping.file_local_id.has_value() ||
1300
344k
            needs_projected_file_type_rebuild(child_mapping)) {
1301
690
            return true;
1302
690
        }
1303
344k
    }
1304
221k
    return false;
1305
222k
}
1306
1307
static std::optional<size_t> file_child_ordinal_in_scan_type(const ColumnMapping& mapping,
1308
309k
                                                             const ColumnMapping& child_mapping) {
1309
309k
    if (!child_mapping.file_local_id.has_value()) {
1310
279
        return std::nullopt;
1311
279
    }
1312
309k
    const auto& file_children = !mapping.projected_file_children.empty()
1313
309k
                                        ? mapping.projected_file_children
1314
18.4E
                                        : mapping.original_file_children;
1315
437k
    const auto child_it = std::ranges::find_if(file_children, [&](const ColumnDefinition& child) {
1316
437k
        return child.file_local_id() == *child_mapping.file_local_id;
1317
437k
    });
1318
309k
    if (child_it == file_children.end()) {
1319
0
        return std::nullopt;
1320
0
    }
1321
309k
    return static_cast<size_t>(std::distance(file_children.begin(), child_it));
1322
309k
}
1323
1324
1.29M
static bool needs_complex_rematerialize(const ColumnMapping& mapping) {
1325
1.29M
    if (mapping.child_mappings.empty()) {
1326
1.08M
        return false;
1327
1.08M
    }
1328
214k
    if (mapping.table_type == nullptr || mapping.file_type == nullptr ||
1329
214k
        !mapping.table_type->equals(*mapping.file_type)) {
1330
11.8k
        return true;
1331
11.8k
    }
1332
509k
    for (size_t table_child_idx = 0; table_child_idx < mapping.child_mappings.size();
1333
309k
         ++table_child_idx) {
1334
309k
        const auto& child_mapping = mapping.child_mappings[table_child_idx];
1335
309k
        const auto file_child_idx = file_child_ordinal_in_scan_type(mapping, child_mapping);
1336
309k
        if (!file_child_idx.has_value() || *file_child_idx != table_child_idx ||
1337
309k
            needs_complex_rematerialize(child_mapping) ||
1338
309k
            (child_mapping.table_type != nullptr && child_mapping.file_type != nullptr &&
1339
307k
             !child_mapping.table_type->equals(*child_mapping.file_type))) {
1340
2.02k
            return true;
1341
2.02k
        }
1342
309k
    }
1343
199k
    return false;
1344
201k
}
1345
1346
1.03M
static bool mapping_can_use_file_column_directly(const ColumnMapping& mapping) {
1347
1.03M
    if (mapping.table_type == nullptr || mapping.file_type == nullptr) {
1348
0
        return false;
1349
0
    }
1350
1.03M
    const auto table_type = remove_nullable(mapping.table_type);
1351
1.03M
    const auto file_type = remove_nullable(mapping.file_type);
1352
1.03M
    const bool same_timestamptz_with_different_scale =
1353
1.03M
            table_type->get_primitive_type() == TYPE_TIMESTAMPTZ &&
1354
1.03M
            file_type->get_primitive_type() == TYPE_TIMESTAMPTZ;
1355
1.03M
    if (!mapping.table_type->equals(*mapping.file_type) && !same_timestamptz_with_different_scale) {
1356
65.7k
        return false;
1357
65.7k
    }
1358
965k
    return !needs_complex_rematerialize(mapping);
1359
1.03M
}
1360
1361
1.28M
static bool type_contains_varbinary(const DataTypePtr& type) {
1362
1.28M
    DORIS_CHECK(type != nullptr);
1363
1.28M
    const auto nested_type = remove_nullable(type);
1364
1.28M
    switch (nested_type->get_primitive_type()) {
1365
1.87k
    case TYPE_VARBINARY:
1366
1.87k
        return true;
1367
105k
    case TYPE_ARRAY:
1368
105k
        return type_contains_varbinary(
1369
105k
                assert_cast<const DataTypeArray&>(*nested_type).get_nested_type());
1370
82.4k
    case TYPE_MAP: {
1371
82.4k
        const auto& map_type = assert_cast<const DataTypeMap&>(*nested_type);
1372
82.4k
        return type_contains_varbinary(map_type.get_key_type()) ||
1373
82.4k
               type_contains_varbinary(map_type.get_value_type());
1374
0
    }
1375
83.8k
    case TYPE_STRUCT:
1376
83.8k
        return std::ranges::any_of(
1377
83.8k
                assert_cast<const DataTypeStruct&>(*nested_type).get_elements(),
1378
165k
                [](const DataTypePtr& child_type) { return type_contains_varbinary(child_type); });
1379
1.01M
    default:
1380
1.01M
        return false;
1381
1.28M
    }
1382
1.28M
}
1383
1384
851k
static FilterConversionType direct_filter_conversion(const ColumnMapping& mapping) {
1385
851k
    DORIS_CHECK(mapping.table_type != nullptr);
1386
851k
    DORIS_CHECK(mapping.file_type != nullptr);
1387
    // FileScanOperator deliberately keeps VARBINARY predicates above external readers. Their
1388
    // physical binary representations are not uniformly supported by reader-side expression and
1389
    // metadata filtering, so localizing a late runtime filter here can incorrectly reject rows.
1390
    // Apply the same rule to a complex root because generic array/map/struct expressions rewrite
1391
    // the root slot and can otherwise expose a nested VARBINARY child to the reader.
1392
851k
    if (type_contains_varbinary(mapping.table_type)) {
1393
1.87k
        return FilterConversionType::FINALIZE_ONLY;
1394
1.87k
    }
1395
849k
    const auto table_type = remove_nullable(mapping.table_type);
1396
849k
    const auto file_type = remove_nullable(mapping.file_type);
1397
    // TIMESTAMPTZ scale mismatch is intentionally materialized as pass-through: a SQL cast rounds
1398
    // fractional seconds. A file-local cast would therefore filter different instants from the
1399
    // scanner-level predicate evaluated on the pass-through value.
1400
849k
    if (table_type->get_primitive_type() == TYPE_TIMESTAMPTZ &&
1401
849k
        file_type->get_primitive_type() == TYPE_TIMESTAMPTZ &&
1402
849k
        !mapping.table_type->equals(*mapping.file_type)) {
1403
1
        return FilterConversionType::FINALIZE_ONLY;
1404
1
    }
1405
849k
    return mapping.is_trivial ? FilterConversionType::COPY_DIRECTLY
1406
849k
                              : FilterConversionType::CAST_FILTER;
1407
849k
}
1408
1409
19.7k
static FilterConversionType projected_filter_conversion(const ColumnMapping& mapping) {
1410
19.7k
    const auto conversion = direct_filter_conversion(mapping);
1411
19.7k
    return !mapping.is_trivial && conversion != FilterConversionType::FINALIZE_ONLY
1412
19.7k
                   ? FilterConversionType::READER_EXPRESSION
1413
19.7k
                   : conversion;
1414
19.7k
}
1415
1416
static const ColumnDefinition* find_file_child_for_mapping(const ColumnDefinition& table_child,
1417
                                                           const ColumnDefinition& file_parent,
1418
                                                           TableColumnMappingMode mode,
1419
                                                           size_t table_child_idx,
1420
332k
                                                           bool allow_ordinal_fallback) {
1421
332k
    const auto file_parent_type = remove_nullable(file_parent.type)->get_primitive_type();
1422
332k
    switch (file_parent_type) {
1423
88.7k
    case TYPE_ARRAY:
1424
88.7k
        DORIS_CHECK(file_parent.children.size() == 1);
1425
88.7k
        return &file_parent.children[0];
1426
149k
    case TYPE_MAP:
1427
149k
        DORIS_CHECK(file_parent.children.size() == 2);
1428
149k
        if (table_child.name == "key") {
1429
74.5k
            return &file_parent.children[0];
1430
74.5k
        }
1431
74.5k
        if (table_child.name == "value") {
1432
74.5k
            return &file_parent.children[1];
1433
74.5k
        }
1434
18.4E
        if (table_child.local_id == 0 || table_child.local_id == 1) {
1435
0
            return &file_parent.children[table_child.local_id];
1436
0
        }
1437
18.4E
        return nullptr;
1438
94.4k
    default:
1439
        // Hive BY_INDEX is a top-level column matching rule. Once a complex root is selected by
1440
        // file position, nested struct children follow Hive reader's historical name matching
1441
        // semantics; their integer identifiers can be field ids, not file positions.
1442
94.4k
        const auto nested_mode =
1443
94.4k
                mode == TableColumnMappingMode::BY_INDEX ? TableColumnMappingMode::BY_NAME : mode;
1444
94.4k
        if (const auto* file_child =
1445
94.4k
                    matcher_for_mode(nested_mode).find(table_child, file_parent.children);
1446
94.4k
            file_child != nullptr) {
1447
88.7k
            return file_child;
1448
88.7k
        }
1449
5.64k
        if (allow_ordinal_fallback && mode == TableColumnMappingMode::BY_FIELD_ID &&
1450
5.64k
            !table_child.has_identifier_field_id()) {
1451
            // Synthetic children are derived from the table DataType when nested ColumnDefinition
1452
            // metadata has been pruned away. They do not carry Iceberg field ids, so try a name
1453
            // match before falling back to ordinal order. Example:
1454
            //   table value type: Struct(age, full_name, gender)
1455
            //   old file value:   Struct(name, age)
1456
            // Name matching keeps `age -> age`; the later unused-child fallback can then map the
1457
            // renamed `full_name -> name` instead of consuming `age` twice.
1458
3
            if (const auto* file_child = NameMatcher().find(table_child, file_parent.children);
1459
3
                file_child != nullptr) {
1460
1
                return file_child;
1461
1
            }
1462
3
        }
1463
        // Some callers only carry the full complex DataType for a projected table column, without
1464
        // expanded nested ColumnDefinitions. In that case we can still preserve full materialization
1465
        // by walking table/file struct fields by ordinal. This is a fallback only: explicit
1466
        // ColumnDefinition children keep using the requested table-format matching rule, which is
1467
        // required for precise schema evolution.
1468
5.64k
        if (allow_ordinal_fallback && table_child_idx < file_parent.children.size()) {
1469
3
            return &file_parent.children[table_child_idx];
1470
3
        }
1471
5.63k
        return nullptr;
1472
332k
    }
1473
332k
}
1474
1475
static ColumnDefinition synthetic_child_definition(const std::string& name, DataTypePtr type,
1476
94.3k
                                                   int32_t local_id) {
1477
94.3k
    ColumnDefinition child;
1478
94.3k
    child.identifier = Field::create_field<TYPE_STRING>(name);
1479
94.3k
    child.local_id = local_id;
1480
94.3k
    child.name = name;
1481
94.3k
    child.type = std::move(type);
1482
94.3k
    return child;
1483
94.3k
}
1484
1485
static std::vector<ColumnDefinition> synthesize_complex_children_from_type(
1486
46.9k
        const DataTypePtr& type) {
1487
46.9k
    std::vector<ColumnDefinition> children;
1488
46.9k
    if (type == nullptr) {
1489
0
        return children;
1490
0
    }
1491
46.9k
    const auto nested_type = remove_nullable(type);
1492
46.9k
    switch (nested_type->get_primitive_type()) {
1493
0
    case TYPE_ARRAY: {
1494
0
        const auto* array_type = assert_cast<const DataTypeArray*>(nested_type.get());
1495
0
        children.push_back(synthetic_child_definition("element", array_type->get_nested_type(), 0));
1496
0
        break;
1497
0
    }
1498
1
    case TYPE_MAP: {
1499
1
        const auto* map_type = assert_cast<const DataTypeMap*>(nested_type.get());
1500
1
        children.push_back(synthetic_child_definition("key", map_type->get_key_type(), 0));
1501
1
        children.push_back(synthetic_child_definition("value", map_type->get_value_type(), 1));
1502
1
        break;
1503
0
    }
1504
46.9k
    case TYPE_STRUCT: {
1505
46.9k
        const auto* struct_type = assert_cast<const DataTypeStruct*>(nested_type.get());
1506
46.9k
        children.reserve(struct_type->get_elements().size());
1507
141k
        for (size_t idx = 0; idx < struct_type->get_elements().size(); ++idx) {
1508
94.3k
            children.push_back(synthetic_child_definition(struct_type->get_element_name(idx),
1509
94.3k
                                                          struct_type->get_element(idx),
1510
94.3k
                                                          cast_set<int32_t>(idx)));
1511
94.3k
        }
1512
46.9k
        break;
1513
0
    }
1514
0
    default:
1515
0
        break;
1516
46.9k
    }
1517
46.9k
    return children;
1518
46.9k
}
1519
1520
static void align_struct_child_types_with_parent(const DataTypePtr& parent_type,
1521
46.9k
                                                 std::vector<ColumnDefinition>& children) {
1522
46.9k
    const auto nested_parent_type = remove_nullable(parent_type);
1523
46.9k
    DORIS_CHECK(nested_parent_type->get_primitive_type() == TYPE_STRUCT);
1524
46.9k
    const auto type_children = synthesize_complex_children_from_type(parent_type);
1525
94.4k
    for (auto& child : children) {
1526
94.4k
        const auto type_child = std::ranges::find_if(
1527
178k
                type_children, [&](const auto& candidate) { return candidate.name == child.name; });
1528
94.4k
        DORIS_CHECK(type_child != type_children.end())
1529
12
                << "Complex child '" << child.name
1530
12
                << "' is absent from its parent table type: " << parent_type->get_name();
1531
        // The parent DataType is the authoritative output contract. Nested schema descriptors can
1532
        // omit child nullability even though the parent struct still declares Nullable(String).
1533
        // For example, the Iceberg full-schema-change case maps nullable `location` to `city`, but
1534
        // its child descriptor carries String. Keeping String here makes rematerialization strip
1535
        // the child's null map and creates Struct(String) under a Struct(Nullable(String)) type.
1536
94.4k
        child.type = type_child->type;
1537
94.4k
    }
1538
46.9k
}
1539
1540
static bool has_table_child_named(const std::vector<ColumnDefinition>& children,
1541
10.5k
                                  std::string_view name) {
1542
15.8k
    return std::ranges::any_of(children, [&](const ColumnDefinition& child) {
1543
15.8k
        return std::string_view(child.name) == name;
1544
15.8k
    });
1545
10.5k
}
1546
1547
static void complete_required_complex_children_from_type(const DataTypePtr& type,
1548
19.1k
                                                         std::vector<ColumnDefinition>& children) {
1549
19.1k
    if (type == nullptr) {
1550
0
        return;
1551
0
    }
1552
19.1k
    const auto nested_type = remove_nullable(type);
1553
19.1k
    switch (nested_type->get_primitive_type()) {
1554
5.29k
    case TYPE_MAP: {
1555
5.29k
        const auto* map_type = assert_cast<const DataTypeMap*>(nested_type.get());
1556
        // MAP key/value are structural children, not independently materializable table fields.
1557
        // A key-only projection can still be attached to a whole-map output root, for example:
1558
        //   SELECT * FROM t WHERE ARRAY_CONTAINS(MAP_KEYS(new_map_column), 'person5')
1559
        //
1560
        // In that shape the scanner keeps the value stream readable, but the table projection can
1561
        // carry only the key child. Add the missing value child so recursive mapping can evolve the
1562
        // value type instead of letting TableReader cast old/new value structs directly.
1563
5.29k
        if (has_table_child_named(children, "key") && !has_table_child_named(children, "value")) {
1564
2
            children.push_back(synthetic_child_definition("value", map_type->get_value_type(), 1));
1565
2
        }
1566
5.29k
        break;
1567
0
    }
1568
5.31k
    case TYPE_ARRAY:
1569
        // ARRAY has only one required structural child (`element`), so a non-empty projection is
1570
        // already rooted at the element path.
1571
5.31k
        break;
1572
8.51k
    case TYPE_STRUCT:
1573
        // STRUCT children are real fields and must remain prunable. Completing missing struct
1574
        // fields here would turn `SELECT s.a` into a full-struct read and undo nested projection.
1575
8.51k
        break;
1576
0
    default:
1577
0
        break;
1578
19.1k
    }
1579
19.1k
}
1580
1581
struct PreparedTableChildren {
1582
    std::vector<ColumnDefinition> children;
1583
    bool synthesized_from_type = false;
1584
};
1585
1586
static PreparedTableChildren prepare_table_children_for_mapping(
1587
832k
        const ColumnDefinition& table_column, const DataTypePtr& file_type) {
1588
832k
    PreparedTableChildren prepared {.children = table_column.children};
1589
832k
    const auto nested_table_type = remove_nullable(table_column.type);
1590
1591
    // Some scan paths, especially SELECT *, only carry the complete complex DataType for a table
1592
    // column and leave ColumnDefinition::children empty. Synthesize the hierarchy so recursive
1593
    // mapping can evolve nested fields instead of falling back to an invalid whole-column cast.
1594
832k
    prepared.synthesized_from_type = prepared.children.empty() &&
1595
832k
                                     is_complex_type(nested_table_type->get_primitive_type()) &&
1596
832k
                                     !table_column.type->equals(*file_type);
1597
832k
    if (prepared.synthesized_from_type) {
1598
5
        prepared.children = synthesize_complex_children_from_type(table_column.type);
1599
832k
    } else if (!prepared.children.empty() && !table_column.type->equals(*file_type)) {
1600
19.1k
        complete_required_complex_children_from_type(table_column.type, prepared.children);
1601
19.1k
    }
1602
1603
832k
    if (!prepared.children.empty() && nested_table_type->get_primitive_type() == TYPE_STRUCT) {
1604
        // Struct children are table fields, so the parent Struct type is authoritative for their
1605
        // nullability. ARRAY and MAP children are format-level structural wrappers and keep the
1606
        // descriptor types used by their recursive mappings.
1607
46.9k
        align_struct_child_types_with_parent(table_column.type, prepared.children);
1608
46.9k
    }
1609
832k
    return prepared;
1610
832k
}
1611
1612
210k
static Status validate_file_schema_children(const ColumnDefinition& file_field) {
1613
210k
    if (file_field.type == nullptr) {
1614
0
        return Status::InternalError("File column '{}' has null type", file_field.name);
1615
0
    }
1616
210k
    const auto nested_type = remove_nullable(file_field.type);
1617
210k
    size_t expected_children = 0;
1618
210k
    bool complex_with_fixed_children = true;
1619
210k
    switch (nested_type->get_primitive_type()) {
1620
88.7k
    case TYPE_ARRAY:
1621
88.7k
        expected_children = 1;
1622
88.7k
        break;
1623
74.5k
    case TYPE_MAP:
1624
74.5k
        expected_children = 2;
1625
74.5k
        break;
1626
46.9k
    case TYPE_STRUCT:
1627
46.9k
        expected_children =
1628
46.9k
                assert_cast<const DataTypeStruct*>(nested_type.get())->get_elements().size();
1629
46.9k
        break;
1630
0
    default:
1631
0
        complex_with_fixed_children = false;
1632
0
        break;
1633
210k
    }
1634
210k
    if (!complex_with_fixed_children || file_field.children.size() == expected_children) {
1635
210k
        return Status::OK();
1636
210k
    }
1637
5
    return Status::InternalError(
1638
5
            "Malformed complex file schema for column '{}': type={}, expected_children={}, "
1639
5
            "actual_children={}",
1640
5
            file_field.name, file_field.type->get_name(), expected_children,
1641
5
            file_field.children.size());
1642
210k
}
1643
1644
450k
static bool has_projected_file_children(const ColumnMapping& mapping) {
1645
450k
    if (mapping.original_file_children.empty() || mapping.projected_file_children.empty()) {
1646
368k
        return false;
1647
368k
    }
1648
81.3k
    if (mapping.original_file_children.size() != mapping.projected_file_children.size()) {
1649
5.82k
        return true;
1650
5.82k
    }
1651
203k
    for (size_t idx = 0; idx < mapping.original_file_children.size(); ++idx) {
1652
127k
        if (mapping.original_file_children[idx].file_local_id() !=
1653
127k
            mapping.projected_file_children[idx].file_local_id()) {
1654
0
            return true;
1655
0
        }
1656
127k
    }
1657
75.5k
    return false;
1658
75.5k
}
1659
1660
450k
static bool needs_nested_file_projection(const ColumnMapping& mapping) {
1661
450k
    if (has_projected_file_children(mapping)) {
1662
        // Return True if the projected child column is missing / re-ordered
1663
5.82k
        return true;
1664
5.82k
    }
1665
444k
    return std::ranges::any_of(mapping.child_mappings, [](const ColumnMapping& child_mapping) {
1666
129k
        return needs_nested_file_projection(child_mapping);
1667
129k
    });
1668
450k
}
1669
1670
static Status build_complex_projection(const ColumnMapping& mapping, LocalColumnIndex* projection);
1671
1672
// Build the projected file children/type according to the pruned complex projection. For example,
1673
// if we have a struct column `s` with children `id` and `name`, and the projection only keeps
1674
// `s.name`, then the file reader should expose `STRUCT<name ...>`.
1675
static Status rebuild_projected_file_children_and_type(
1676
        const DataTypePtr& file_type, const std::vector<ColumnDefinition>& original_file_children,
1677
        const std::vector<ColumnMapping>& child_mappings,
1678
19.7k
        std::vector<ColumnDefinition>* projected_file_children, DataTypePtr* projected_type) {
1679
19.7k
    DORIS_CHECK(file_type != nullptr);
1680
19.7k
    DORIS_CHECK(projected_file_children != nullptr);
1681
19.7k
    DORIS_CHECK(projected_type != nullptr);
1682
19.7k
    ColumnDefinition field;
1683
19.7k
    field.type = file_type;
1684
19.7k
    field.children = original_file_children;
1685
19.7k
    LocalColumnIndex projection = LocalColumnIndex::partial_local(-1);
1686
19.7k
    projection.children.reserve(child_mappings.size());
1687
30.3k
    for (const auto* child_mapping : present_child_mappings_in_file_order(child_mappings)) {
1688
30.3k
        DORIS_CHECK(child_mapping->file_local_id.has_value());
1689
30.3k
        LocalColumnIndex child_projection;
1690
30.3k
        RETURN_IF_ERROR(build_complex_projection(*child_mapping, &child_projection));
1691
30.3k
        projection.children.push_back(std::move(child_projection));
1692
30.3k
    }
1693
1694
19.7k
    ColumnDefinition projected_field;
1695
19.7k
    RETURN_IF_ERROR(project_column_definition(field, projection, &projected_field));
1696
19.7k
    *projected_file_children = std::move(projected_field.children);
1697
19.7k
    *projected_type = std::move(projected_field.type);
1698
19.7k
    return Status::OK();
1699
19.7k
}
1700
1701
// Build the complex column projection according to the ColumnMapping which is re-ordered by the
1702
// file-schema's order.
1703
//
1704
// For MAP, a partial projection represents value-subtree pruning only. The key child is not a
1705
// projected output shape; file readers still read full keys to construct ColumnMap offsets and keep
1706
// key semantics unchanged. If a caller tries to project only/prune the key child, the common schema
1707
// projection helper rejects it.
1708
55.4k
static Status build_complex_projection(const ColumnMapping& mapping, LocalColumnIndex* projection) {
1709
55.4k
    if (projection == nullptr) {
1710
0
        return Status::InvalidArgument("projection is null");
1711
0
    }
1712
55.4k
    DORIS_CHECK(mapping.file_local_id.has_value());
1713
55.4k
    *projection = LocalColumnIndex::local(*mapping.file_local_id);
1714
55.4k
    projection->timestamp_is_adjusted_to_utc = mapping.timestamp_is_adjusted_to_utc;
1715
55.4k
    projection->project_all_children = mapping.child_mappings.empty();
1716
55.4k
    projection->children.clear();
1717
55.4k
    const auto present_children = present_child_mappings_in_file_order(mapping.child_mappings);
1718
55.4k
    if (!projection->project_all_children && present_children.empty()) {
1719
        // All requested table children under this complex node are missing/default-only. The file
1720
        // reader cannot expose an empty complex projection, but TableReader can still rematerialize
1721
        // the table shape from a full file subtree and fill the missing children with defaults.
1722
260
        projection->project_all_children = true;
1723
260
        return Status::OK();
1724
260
    }
1725
55.1k
    for (const auto* child_mapping : present_children) {
1726
19.2k
        LocalColumnIndex child_projection;
1727
19.2k
        RETURN_IF_ERROR(build_complex_projection(*child_mapping, &child_projection));
1728
19.2k
        projection->children.push_back(std::move(child_projection));
1729
19.2k
    }
1730
55.1k
    if (!projection->project_all_children && projection->children.empty()) {
1731
0
        return Status::NotSupported("Projection for complex column {} contains no file children",
1732
0
                                    mapping.file_column_name);
1733
0
    }
1734
55.1k
    return Status::OK();
1735
55.1k
}
1736
1737
330k
static bool has_timestamp_semantics(const ColumnMapping& mapping) {
1738
330k
    return mapping.timestamp_is_adjusted_to_utc.has_value() ||
1739
330k
           std::ranges::any_of(mapping.child_mappings, has_timestamp_semantics);
1740
330k
}
1741
1742
517k
static void attach_timestamp_semantics(const ColumnMapping& mapping, LocalColumnIndex* projection) {
1743
517k
    DORIS_CHECK(projection != nullptr);
1744
517k
    projection->timestamp_is_adjusted_to_utc = mapping.timestamp_is_adjusted_to_utc;
1745
517k
    for (const auto& child_mapping : mapping.child_mappings) {
1746
        // A full projection represents ordinary children implicitly; materialize only paths that
1747
        // carry an override so existing readers still observe an empty children list.
1748
283k
        if (!child_mapping.file_local_id.has_value() || !has_timestamp_semantics(child_mapping)) {
1749
282k
            continue;
1750
282k
        }
1751
854
        auto child_it =
1752
854
                std::ranges::find_if(projection->children, [&](const LocalColumnIndex& child) {
1753
64
                    return child.local_id() == *child_mapping.file_local_id;
1754
64
                });
1755
858
        if (child_it == projection->children.end()) {
1756
858
            projection->children.push_back(LocalColumnIndex::local(*child_mapping.file_local_id));
1757
858
            child_it = std::prev(projection->children.end());
1758
858
        }
1759
854
        attach_timestamp_semantics(child_mapping, &*child_it);
1760
854
    }
1761
517k
}
1762
1763
using FilterProjectionMap = std::map<LocalColumnId, LocalColumnIndex>;
1764
1765
// Update the mapping's file type according to the projection, and determine whether the projection
1766
// is trivial (i.e. the projected file type is the same as the table type, so no need to
1767
// rematerialize the complex value back to table layout after reading from file).
1768
static Status apply_projection_to_mapping_file_type(const LocalColumnIndex& projection,
1769
505k
                                                    ColumnMapping* mapping) {
1770
505k
    DORIS_CHECK(mapping != nullptr);
1771
505k
    if (mapping->original_file_type == nullptr) {
1772
0
        mapping->original_file_type = mapping->file_type;
1773
0
    }
1774
505k
    if (mapping->original_file_type == nullptr ||
1775
505k
        !is_complex_type(remove_nullable(mapping->original_file_type)->get_primitive_type())) {
1776
326k
        return Status::OK();
1777
326k
    }
1778
179k
    ColumnDefinition field;
1779
179k
    field.type = mapping->original_file_type;
1780
179k
    field.children = mapping->original_file_children;
1781
179k
    ColumnDefinition projected_field;
1782
179k
    RETURN_IF_ERROR(project_column_definition(field, projection, &projected_field));
1783
179k
    mapping->file_type = std::move(projected_field.type);
1784
179k
    mapping->projected_file_children = std::move(projected_field.children);
1785
179k
    mapping->is_trivial = mapping_can_use_file_column_directly(*mapping);
1786
179k
    return Status::OK();
1787
179k
}
1788
1789
static Status merge_filter_projection(const FilterProjectionMap* filter_projections,
1790
73.6k
                                      LocalColumnIndex* projection) {
1791
73.6k
    DORIS_CHECK(projection != nullptr);
1792
73.6k
    if (filter_projections == nullptr) {
1793
0
        return Status::OK();
1794
0
    }
1795
73.6k
    const auto filter_projection_it = filter_projections->find(projection->column_id());
1796
73.6k
    if (filter_projection_it == filter_projections->end()) {
1797
72.3k
        return Status::OK();
1798
72.3k
    }
1799
    // Merge predicate-only nested paths into the root projection that is about to be scanned.
1800
    // Example: `SELECT s.a WHERE s.b > 1` first builds the output projection `s -> a` from
1801
    // ColumnMapping, while build_nested_struct_filter_projection_map() records `s -> b`. This merge
1802
    // produces one file scan projection `s -> a,b`.
1803
1.29k
    RETURN_IF_ERROR(merge_local_column_index(projection, filter_projection_it->second));
1804
1.29k
    return Status::OK();
1805
1.29k
}
1806
1807
172
static bool table_root_is_map(const ColumnMapping& mapping) {
1808
172
    if (mapping.table_type == nullptr) {
1809
0
        return false;
1810
0
    }
1811
172
    return remove_nullable(mapping.table_type)->get_primitive_type() == TYPE_MAP;
1812
172
}
1813
1814
static Status add_scan_column(FileScanRequest* file_request, ColumnMapping* mapping,
1815
                              bool is_predicate_column, bool force_full_complex_scan_projection,
1816
516k
                              const FilterProjectionMap* filter_projections = nullptr) {
1817
516k
    const auto file_column_id = LocalColumnId(mapping->file_local_id.value());
1818
516k
    LocalColumnIndex projection = LocalColumnIndex::top_level(file_column_id);
1819
516k
    projection.timestamp_is_adjusted_to_utc = mapping->timestamp_is_adjusted_to_utc;
1820
    // Columnar readers can turn a complex mapping into a nested file projection, but
1821
    // row-oriented readers must scan the full top-level complex field because all children are
1822
    // encoded in the same text cell.
1823
516k
    if (!force_full_complex_scan_projection && needs_nested_file_projection(*mapping)) {
1824
5.82k
        RETURN_IF_ERROR(build_complex_projection(*mapping, &projection));
1825
5.82k
    }
1826
516k
    if (is_predicate_column && !force_full_complex_scan_projection) {
1827
73.6k
        DCHECK(filter_projections != nullptr);
1828
        // If a projected complex root is also used by a predicate, rebuild the predicate scan
1829
        // projection from the output mapping before merging predicate-only children. For
1830
        // `SELECT s.a WHERE s.b > 1`, build_complex_projection() produces `s -> a` and
1831
        // merge_filter_projection() adds `s -> b`, so the predicate column reads both children.
1832
73.6k
        RETURN_IF_ERROR(merge_filter_projection(filter_projections, &projection));
1833
73.6k
    }
1834
516k
    attach_timestamp_semantics(*mapping, &projection);
1835
516k
    FileScanRequestBuilder builder(file_request);
1836
516k
    if (is_predicate_column) {
1837
73.6k
        return builder.add_predicate_column(std::move(projection));
1838
73.6k
    }
1839
442k
    return builder.add_non_predicate_column(std::move(projection));
1840
516k
}
1841
1842
static const LocalColumnIndex* find_scan_projection(
1843
948k
        const std::vector<LocalColumnIndex>& scan_columns, LocalColumnId file_column_id) {
1844
948k
    const auto projection_it =
1845
8.98M
            std::ranges::find_if(scan_columns, [&](const LocalColumnIndex& projection) {
1846
8.98M
                return projection.column_id() == file_column_id;
1847
8.98M
            });
1848
948k
    return projection_it == scan_columns.end() ? nullptr : &*projection_it;
1849
948k
}
1850
1851
// Apply the final scan projection of one root file column back to its ColumnMapping. This updates
1852
// mapping.file_type/projected_file_children from the original file schema to the exact shape that
1853
// FileReader will return.
1854
//
1855
// Example: for `SELECT s.a WHERE s.b > 1`, add_scan_column() keeps only one predicate scan
1856
// projection `s -> a,b`. Applying that projection changes the mapping's file type from the full
1857
// file struct `s<a,b,c>` to the projected file struct `s<a,b>`, so later filter rewrite and
1858
// TableReader final materialization use the same column shape as the file-local block.
1859
static Status apply_scan_projection_to_mapping_file_type(const FileScanRequest& file_request,
1860
506k
                                                         ColumnMapping* mapping) {
1861
506k
    DORIS_CHECK(mapping != nullptr);
1862
506k
    DORIS_CHECK(mapping->file_local_id.has_value());
1863
506k
    const auto file_column_id = LocalColumnId(*mapping->file_local_id);
1864
    // Predicate columns are the actual scan projection when a column is used by row-level filters:
1865
    // add_scan_column() removes the duplicate non-predicate projection in that case.
1866
506k
    const auto* projection = find_scan_projection(file_request.predicate_columns, file_column_id);
1867
506k
    if (projection == nullptr) {
1868
442k
        projection = find_scan_projection(file_request.non_predicate_columns, file_column_id);
1869
442k
    }
1870
506k
    DORIS_CHECK(projection != nullptr);
1871
506k
    return apply_projection_to_mapping_file_type(*projection, mapping);
1872
506k
}
1873
1874
// Build extra scan projections required only by row-level filters on nested struct children.
1875
//
1876
// Example: for `SELECT s.a FROM t WHERE s.b.c > 1`, the output projection may only contain `s.a`,
1877
// but the file reader must also read `s.b.c` to evaluate the predicate. This function collects the
1878
// table-side filter path, resolves it through ColumnMapping first, and records the corresponding
1879
// file-side projection in filter_projections. This keeps renamed fields consistent between the scan
1880
// projection and row-level conjunct rewrite. Example:
1881
//   table filter path: s -> renamed_b -> c
1882
//   old file path:     s -> b -> c
1883
//   recorded path:     s -> b -> c
1884
// When add_scan_column() adds the same root as a predicate column, it rebuilds that root from the
1885
// output mapping, merges this filter-only projection into it, and removes the duplicate
1886
// non-predicate root entry.
1887
static Status build_nested_struct_filter_projection_map(
1888
        const std::vector<TableFilter>& table_filters, const std::vector<ColumnMapping>& mappings,
1889
122k
        FilterProjectionMap* filter_projections) {
1890
122k
    DORIS_CHECK(filter_projections != nullptr);
1891
122k
    filter_projections->clear();
1892
122k
    for (const auto& table_filter : table_filters) {
1893
79.9k
        if (table_filter.conjunct == nullptr) {
1894
0
            continue;
1895
0
        }
1896
        // Collect all nested struct paths in the table filter. For example, for
1897
        // `s.id > 5 AND element_at(s, 'renamed_name') = 'abc'`, collect the table paths
1898
        // `s -> id` and `s -> renamed_name`, then resolve each one to its file-side projection.
1899
79.9k
        std::vector<NestedStructPath> paths;
1900
79.9k
        collect_nested_struct_paths(table_filter.conjunct->root(), &paths);
1901
79.9k
        for (const auto& path : paths) {
1902
5.45k
            auto mapping_it = std::ranges::find_if(mappings, [&](const ColumnMapping& mapping) {
1903
5.45k
                return mapping.global_index == path.root_global_index;
1904
5.45k
            });
1905
1.87k
            if (mapping_it == mappings.end() || !mapping_it->file_local_id.has_value() ||
1906
1.87k
                path.selectors.empty()) {
1907
440
                continue;
1908
440
            }
1909
1910
1.43k
            ResolvedNestedStructPath resolved;
1911
1.43k
            LocalColumnIndex root_projection;
1912
1.43k
            if (!resolve_nested_struct_path_for_file(path, mappings, &resolved)) {
1913
172
                if (!table_root_is_map(*mapping_it)) {
1914
171
                    continue;
1915
171
                }
1916
                // Direct map value filters such as `m.value.a > 1` need the value leaf for row
1917
                // evaluation even when the query only projects another value child. This is only a
1918
                // scan projection fallback; complex map/array expressions are still not rewritten
1919
                // into file-local conjuncts.
1920
1
                LocalColumnIndex child_projection;
1921
1
                RETURN_IF_ERROR(build_file_child_projection_from_schema(
1922
1
                        mapping_it->original_file_children, path.selectors, &child_projection));
1923
1
                if (child_projection.local_id() < 0) {
1924
0
                    continue;
1925
0
                }
1926
1
                root_projection = LocalColumnIndex::partial_local(*mapping_it->file_local_id);
1927
1
                root_projection.children.push_back(std::move(child_projection));
1928
1.26k
            } else {
1929
1.26k
                root_projection = std::move(resolved.file_projection);
1930
1.26k
            }
1931
1.26k
            auto filter_projection_it = filter_projections->find(root_projection.column_id());
1932
1.26k
            if (filter_projection_it == filter_projections->end()) {
1933
1.19k
                filter_projections->emplace(root_projection.column_id(),
1934
1.19k
                                            std::move(root_projection));
1935
1.19k
                continue;
1936
1.19k
            }
1937
68
            RETURN_IF_ERROR(
1938
68
                    merge_local_column_index(&filter_projection_it->second, root_projection));
1939
68
        }
1940
79.9k
    }
1941
122k
    return Status::OK();
1942
122k
}
1943
1944
506k
static void rebuild_projection(ColumnMapping* mapping, LocalIndex block_position) {
1945
506k
    DORIS_CHECK(mapping->file_local_id.has_value());
1946
506k
    if (mapping->is_trivial || needs_complex_rematerialize(*mapping)) {
1947
496k
        mapping->projection = VExprContext::create_shared(VSlotRef::create_shared(
1948
496k
                cast_set<int>(block_position.value()), cast_set<int>(block_position.value()), -1,
1949
496k
                mapping->file_type, mapping->file_column_name));
1950
496k
        return;
1951
496k
    }
1952
1953
9.83k
    auto expr = Cast::create_shared(mapping->table_type);
1954
9.83k
    expr->add_child(VSlotRef::create_shared(cast_set<int>(block_position.value()),
1955
9.83k
                                            cast_set<int>(block_position.value()), -1,
1956
9.83k
                                            mapping->file_type, mapping->file_column_name));
1957
9.83k
    mapping->projection = VExprContext::create_shared(expr);
1958
9.83k
}
1959
1960
// Build file slot rewrite info from the localized filter targets. Only local targets can enter
1961
// file-reader expressions; constant and unset targets stay above the file reader.
1962
static std::map<GlobalIndex, FileSlotRewriteInfo> build_file_slot_rewrite_map(
1963
        const std::vector<ColumnMapping>& mappings,
1964
122k
        const std::map<GlobalIndex, FilterEntry>& filter_entries) {
1965
122k
    std::map<GlobalIndex, FileSlotRewriteInfo> global_to_file_slot;
1966
533k
    for (const auto& mapping : mappings) {
1967
533k
        const auto entry_it = filter_entries.find(mapping.global_index);
1968
533k
        if (entry_it == filter_entries.end() || !entry_it->second.is_local()) {
1969
28.5k
            continue;
1970
28.5k
        }
1971
504k
        DORIS_CHECK(mapping.file_local_id.has_value());
1972
504k
        global_to_file_slot.emplace(
1973
504k
                mapping.global_index,
1974
504k
                FileSlotRewriteInfo {.block_position = entry_it->second.local_index().value(),
1975
504k
                                     .file_type = mapping.file_type,
1976
504k
                                     .table_type = mapping.table_type,
1977
504k
                                     .file_column_name = mapping.file_column_name});
1978
504k
    }
1979
122k
    return global_to_file_slot;
1980
122k
}
1981
1982
Status TableColumnMapper::_create_by_index_mapping(const ColumnDefinition& table_column,
1983
                                                   const std::vector<ColumnDefinition>& file_schema,
1984
33.7k
                                                   ColumnMapping* mapping) {
1985
33.7k
    DORIS_CHECK(mapping != nullptr);
1986
33.7k
    DORIS_CHECK(!table_column.is_partition_key);
1987
1988
    // Key contract: in BY_INDEX mode, `ColumnDefinition::identifier` TYPE_INT is interpreted as the
1989
    // 0-based position of this column inside `file_schema`. FE writes the physical file position
1990
    // of each non-partition projected column into that identifier. This interpretation allows:
1991
    //   - sparse projection: read only a subset of file columns (for example only `_col2`
1992
    //     and `_col4`);
1993
    //   - column reordering: table column order differs from file column order;
1994
    //   - no many-to-one mapping: FE must guarantee that each file position is referenced by at
1995
    //     most one table column.
1996
33.7k
    const auto file_index = table_column.get_identifier_position();
1997
1998
    // Case A: file_index is in range, so build a direct positional mapping.
1999
    // The file column name (for example `_col0`) is intentionally ignored here.
2000
33.7k
    if (file_index >= 0 && static_cast<size_t>(file_index) < file_schema.size()) {
2001
33.1k
        return _create_direct_mapping(table_column, file_schema[static_cast<size_t>(file_index)],
2002
33.1k
                                      mapping);
2003
33.1k
    }
2004
2005
    // Case B: file_index is out of range, which means the file does not contain this column.
2006
    // Route it through the missing-column path used by schema evolution.
2007
531
    if (table_column.default_expr != nullptr) {
2008
531
        _set_constant_mapping(mapping, table_column.default_expr);
2009
531
        return Status::OK();
2010
531
    }
2011
    // Keep the mapping empty (`file_local_id` remains `nullopt`) and let the upper finalize
2012
    // stage fill NULL/default values.
2013
18.4E
    return Status::OK();
2014
527
}
2015
2016
22.9k
void TableColumnMapper::_set_constant_mapping(ColumnMapping* mapping, VExprContextSPtr expr) {
2017
22.9k
    DORIS_CHECK(mapping != nullptr);
2018
22.9k
    DORIS_CHECK(expr != nullptr);
2019
22.9k
    mapping->default_expr = std::move(expr);
2020
22.9k
    mapping->constant_index = _constant_map.add(ConstantEntry {
2021
22.9k
            .global_index = mapping->global_index,
2022
22.9k
            .expr = mapping->default_expr,
2023
22.9k
            .type = mapping->table_type,
2024
22.9k
    });
2025
22.9k
    mapping->filter_conversion = FilterConversionType::CONSTANT;
2026
22.9k
}
2027
2028
Status TableColumnMapper::_create_mapping_for_column(const ColumnDefinition& table_column,
2029
                                                     GlobalIndex global_index,
2030
532k
                                                     ColumnMapping* mapping) {
2031
532k
    DORIS_CHECK(mapping != nullptr);
2032
532k
    *mapping = ColumnMapping {};
2033
532k
    mapping->global_index = global_index;
2034
532k
    mapping->table_column_name = table_column.name;
2035
532k
    mapping->table_type = table_column.type;
2036
    // Row-lineage names are Iceberg metadata contracts, not reserved names in generic Hive,
2037
    // Hudi, or Paimon schemas. Only the Iceberg reader may opt into virtual synthesis.
2038
532k
    const auto row_lineage_type =
2039
532k
            _options.enable_row_lineage_virtual_columns
2040
532k
                    ? row_lineage_virtual_column_type(table_column, _options.mode)
2041
532k
                    : TableVirtualColumnType::INVALID;
2042
532k
    if (const auto* partition_value = find_partition_value(table_column, _partition_values);
2043
532k
        table_column.is_partition_key && partition_value != nullptr) {
2044
        // Partition values are split constants and must take precedence over defaults.
2045
15.9k
        _set_constant_mapping(mapping, VExprContext::create_shared(VLiteral::create_shared(
2046
15.9k
                                               mapping->table_type, *partition_value)));
2047
516k
    } else if (_options.mode == TableColumnMappingMode::BY_INDEX &&
2048
516k
               !table_column.is_partition_key && table_column.has_identifier_field_id()) {
2049
        // BY_INDEX interprets ColumnDefinition::identifier as physical file position.
2050
33.7k
        RETURN_IF_ERROR(_create_by_index_mapping(table_column, _file_schema, mapping));
2051
483k
    } else if (const auto* file_field = _find_file_field(table_column, _file_schema)) {
2052
        // Normal physical file column mapping.
2053
473k
        RETURN_IF_ERROR(_create_direct_mapping(table_column, *file_field, mapping));
2054
473k
        if (row_lineage_type != TableVirtualColumnType::INVALID) {
2055
            // Iceberg v3 rewritten files may physically contain row lineage metadata fields.
2056
            // File non-null values must be preserved, while file NULLs still inherit from data file
2057
            // metadata in IcebergTableReader. Therefore the mapping has a real file source plus a
2058
            // virtual post-materialization step, and filters must wait for finalize output.
2059
666
            mapping->virtual_column_type = row_lineage_type;
2060
666
            mapping->filter_conversion = FilterConversionType::FINALIZE_ONLY;
2061
666
        }
2062
473k
    } else if (row_lineage_type != TableVirtualColumnType::INVALID) {
2063
        // Iceberg row lineage metadata fields are optional in data files. Missing fields are exposed
2064
        // as all-NULL table columns first; IcebergTableReader fills inherited values only when the
2065
        // split carries first_row_id / last_updated_sequence_number metadata.
2066
        // FE may attach a default_expr to these hidden metadata columns, but the Iceberg v3
2067
        // inheritance rule must take precedence over the generic missing-column default path.
2068
3.20k
        mapping->virtual_column_type = row_lineage_type;
2069
6.80k
    } else if (table_column.name == BeConsts::ICEBERG_ROWID_COL) {
2070
        // Doris internal Iceberg row locator is never a physical Iceberg data column. It is built
2071
        // from file path, row position and partition metadata for delete/update/merge.
2072
560
        mapping->virtual_column_type = TableVirtualColumnType::ICEBERG_ROWID;
2073
6.43k
    } else if (table_column.default_expr != nullptr) {
2074
        // Table-format readers build typed default expressions before mapping. Keep that typed
2075
        // expression authoritative over the raw transport metadata, which cannot represent complex
2076
        // defaults safely in this table-format-neutral layer.
2077
6.43k
        _set_constant_mapping(mapping, table_column.default_expr);
2078
18.4E
    } else if (table_column.initial_default_value.has_value()) {
2079
0
        return Status::InvalidArgument(
2080
0
                "Missing typed initial-default expression for table field '{}'", table_column.name);
2081
18.4E
    } else if (_options.reject_missing_required_field && table_column.is_optional.has_value() &&
2082
18.4E
               !*table_column.is_optional) {
2083
1
        return Status::InvalidArgument("Missing required field: {}", table_column.name);
2084
18.4E
    } else {
2085
18.4E
        if (table_column.is_partition_key) {
2086
0
            return Status::InvalidArgument(
2087
0
                    "Table column '{}' (global_index={}) does not have a matching partition value",
2088
0
                    table_column.name, mapping->global_index.value());
2089
0
        }
2090
18.4E
    }
2091
532k
    return Status::OK();
2092
532k
}
2093
2094
Status TableColumnMapper::_create_hidden_filter_mapping(const ColumnDefinition& table_column,
2095
                                                        GlobalIndex global_index,
2096
2
                                                        ColumnMapping* mapping) {
2097
2
    auto status = _create_mapping_for_column(table_column, global_index, mapping);
2098
2
    if (mapping->file_local_id.has_value() || mapping->constant_index.has_value() ||
2099
2
        mapping->virtual_column_type != TableVirtualColumnType::INVALID) {
2100
0
        return Status::OK();
2101
0
    }
2102
2
    if (_options.mode == TableColumnMappingMode::BY_NAME) {
2103
0
        return status;
2104
0
    }
2105
2106
    // Predicate-only slot refs carry the table name/type but do not carry the table-format field
2107
    // id used by BY_FIELD_ID or the file position used by BY_INDEX. Use a name fallback only for
2108
    // hidden filter localization; projected columns still obey the requested mapping mode.
2109
2
    const auto* file_field =
2110
2
            matcher_for_mode(TableColumnMappingMode::BY_NAME).find(table_column, _file_schema);
2111
2
    if (file_field == nullptr) {
2112
0
        return status;
2113
0
    }
2114
2
    ColumnMapping fallback_mapping;
2115
2
    fallback_mapping.global_index = global_index;
2116
2
    fallback_mapping.table_column_name = table_column.name;
2117
2
    fallback_mapping.table_type = table_column.type;
2118
2
    RETURN_IF_ERROR(_create_direct_mapping(table_column, *file_field, &fallback_mapping));
2119
2
    *mapping = std::move(fallback_mapping);
2120
2
    return Status::OK();
2121
2
}
2122
2123
Status TableColumnMapper::_build_hidden_filter_mappings(
2124
122k
        const std::vector<TableFilter>& table_filters) {
2125
122k
    _hidden_mappings.clear();
2126
2127
122k
    std::map<GlobalIndex, ColumnDefinition> filter_columns;
2128
122k
    for (const auto& table_filter : table_filters) {
2129
79.8k
        if (table_filter.conjunct != nullptr) {
2130
79.8k
            collect_top_level_slot_columns(table_filter.conjunct->root(), &filter_columns);
2131
79.8k
        }
2132
79.8k
    }
2133
2134
122k
    for (const auto& [global_index, table_column] : filter_columns) {
2135
70.2k
        if (_find_mapping(global_index) != nullptr) {
2136
            // Ignore columns that are already mapped by the projected columns
2137
70.2k
            continue;
2138
70.2k
        }
2139
6
        ColumnMapping mapping;
2140
6
        RETURN_IF_ERROR(_create_hidden_filter_mapping(table_column, global_index, &mapping));
2141
6
        if (mapping.file_local_id.has_value() || mapping.constant_index.has_value() ||
2142
6
            mapping.virtual_column_type != TableVirtualColumnType::INVALID) {
2143
2
            _hidden_mappings.push_back(std::move(mapping));
2144
2
        }
2145
6
    }
2146
122k
    return Status::OK();
2147
122k
}
2148
2149
Status TableColumnMapper::create_mapping(const std::vector<ColumnDefinition>& projected_columns,
2150
                                         const std::map<std::string, Field>& partition_values,
2151
122k
                                         const std::vector<ColumnDefinition>& file_schema) {
2152
122k
    clear();
2153
122k
    _partition_values = partition_values;
2154
122k
    _file_schema = file_schema;
2155
655k
    for (size_t column_idx = 0; column_idx < projected_columns.size(); ++column_idx) {
2156
533k
        ColumnMapping mapping;
2157
533k
        RETURN_IF_ERROR(_create_mapping_for_column(projected_columns[column_idx],
2158
533k
                                                   GlobalIndex(column_idx), &mapping));
2159
533k
        _mappings.push_back(std::move(mapping));
2160
533k
    }
2161
122k
    return Status::OK();
2162
122k
}
2163
2164
367k
std::vector<ColumnMapping> TableColumnMapper::_filter_visible_mappings() const {
2165
367k
    std::vector<ColumnMapping> mappings;
2166
367k
    mappings.reserve(_mappings.size() + _hidden_mappings.size());
2167
367k
    mappings.insert(mappings.end(), _mappings.begin(), _mappings.end());
2168
367k
    mappings.insert(mappings.end(), _hidden_mappings.begin(), _hidden_mappings.end());
2169
367k
    return mappings;
2170
367k
}
2171
2172
122k
Status TableColumnMapper::_build_filter_entries(const FileScanRequest& file_request) {
2173
122k
    _filter_entries.clear();
2174
122k
    const auto mappings = _filter_visible_mappings();
2175
533k
    for (const auto& mapping : mappings) {
2176
533k
        FilterEntry entry;
2177
533k
        if (mapping.constant_index.has_value()) {
2178
22.9k
            entry = FilterEntry::constant(*mapping.constant_index);
2179
510k
        } else if (mapping.file_local_id.has_value() &&
2180
510k
                   filter_conversion_has_local_source(mapping.filter_conversion)) {
2181
504k
            const auto local_position_it =
2182
504k
                    file_request.local_positions.find(LocalColumnId(*mapping.file_local_id));
2183
504k
            if (local_position_it != file_request.local_positions.end()) {
2184
504k
                entry = FilterEntry::local(local_position_it->second);
2185
504k
            }
2186
504k
        }
2187
533k
        _filter_entries.emplace(mapping.global_index, entry);
2188
533k
    }
2189
122k
    return Status::OK();
2190
122k
}
2191
2192
Status TableColumnMapper::create_scan_request(
2193
        const std::vector<TableFilter>& table_filters,
2194
        const std::vector<ColumnDefinition>& projected_columns, FileScanRequest* file_request,
2195
        RuntimeState* runtime_state,
2196
122k
        const std::map<LocalColumnId, LocalIndex>* fixed_local_positions) {
2197
    // FileReader evaluates expressions against a file-local block. This mapper owns the
2198
    // table-column to file-column conversion, so it also owns the file-local block positions.
2199
122k
    file_request->predicate_columns.clear();
2200
122k
    file_request->non_predicate_columns.clear();
2201
122k
    file_request->predicate_only_columns.clear();
2202
122k
    file_request->local_positions.clear();
2203
122k
    if (fixed_local_positions != nullptr) {
2204
        // A refreshed predicate may promote a lazy column, but the active split's block slots are
2205
        // immutable. Seed their positions before rebuilding expressions so every rewritten SlotRef
2206
        // continues to address the same physical column.
2207
37
        file_request->local_positions = *fixed_local_positions;
2208
37
    }
2209
122k
    file_request->conjuncts.clear();
2210
122k
    file_request->delete_conjuncts.clear();
2211
122k
    _filter_entries.clear();
2212
    // 1. Build referenced non-predicate columns
2213
655k
    for (size_t column_idx = 0; column_idx < projected_columns.size(); ++column_idx) {
2214
533k
        const auto global_index = GlobalIndex(column_idx);
2215
533k
        auto* mapping = _find_mapping(global_index);
2216
533k
        if (mapping != nullptr && mapping->file_local_id.has_value()) {
2217
            // A file column can be read lazily as a non-predicate column only when it is not used
2218
            // by row-level expression filters.
2219
506k
            bool used_by_filter = false;
2220
506k
            for (const auto& table_filter : table_filters) {
2221
142k
                const auto& global_indices = table_filter.global_indices;
2222
142k
                if (std::find(global_indices.begin(), global_indices.end(), global_index) !=
2223
142k
                            global_indices.end() &&
2224
142k
                    filter_conversion_has_local_source(mapping->filter_conversion)) {
2225
64.1k
                    used_by_filter = true;
2226
64.1k
                    break;
2227
64.1k
                }
2228
142k
            }
2229
506k
            if (!used_by_filter || !enable_lazy_materialization()) {
2230
442k
                RETURN_IF_ERROR(add_scan_column(file_request, mapping, false,
2231
442k
                                                force_full_complex_scan_projection()));
2232
442k
            }
2233
506k
        }
2234
533k
    }
2235
    // 2. Build referenced predicate columns
2236
    // Hidden filter mappings must be built before localizing filters, so that they can be localized together with visible mappings and referenced by localized filter expressions.
2237
122k
    RETURN_IF_ERROR(_build_hidden_filter_mappings(table_filters));
2238
122k
    RETURN_IF_ERROR(localize_filters(table_filters, file_request, runtime_state));
2239
122k
    for (const auto& mapping : _hidden_mappings) {
2240
2
        if (!mapping.file_local_id.has_value()) {
2241
0
            continue;
2242
0
        }
2243
2
        const auto local_id = LocalColumnId(*mapping.file_local_id);
2244
2
        const bool is_visible_output =
2245
2
                std::ranges::any_of(_mappings, [local_id](const ColumnMapping& visible_mapping) {
2246
2
                    return visible_mapping.file_local_id.has_value() &&
2247
2
                           LocalColumnId(*visible_mapping.file_local_id) == local_id;
2248
2
                });
2249
2
        if (is_visible_output) {
2250
0
            continue;
2251
0
        }
2252
        // File-local filtering is an optimization; Scanner still evaluates the original
2253
        // table-level conjunct after TableReader returns. Only truly hidden mappings are absent
2254
        // from that scanner-visible block and may safely discard their payload here.
2255
2
        if (std::ranges::any_of(file_request->predicate_columns,
2256
2
                                [local_id](const LocalColumnIndex& projection) {
2257
2
                                    return projection.column_id() == local_id;
2258
2
                                }) &&
2259
2
            !file_request->is_predicate_only(local_id)) {
2260
2
            file_request->predicate_only_columns.push_back(local_id);
2261
2
        }
2262
2
    }
2263
    // 3. Rebuild output projection expressions for projected columns. localize_filters() has
2264
    // already applied the final scan projection to mapping.file_type/projected_file_children before
2265
    // rewriting filter expressions.
2266
533k
    for (auto& mapping : _mappings) {
2267
533k
        if (!mapping.file_local_id.has_value()) {
2268
26.6k
            continue;
2269
26.6k
        }
2270
506k
        auto position_it =
2271
506k
                file_request->local_positions.find(LocalColumnId(*mapping.file_local_id));
2272
506k
        DORIS_CHECK(position_it != file_request->local_positions.end())
2273
184
                << file_request->local_positions.size() << " " << *mapping.file_local_id << " "
2274
184
                << mapping.file_column_name;
2275
506k
        rebuild_projection(&mapping, position_it->second);
2276
506k
    }
2277
122k
    return Status::OK();
2278
122k
}
2279
2280
755k
ColumnMapping* TableColumnMapper::_find_mapping(GlobalIndex global_index) {
2281
9.46M
    for (auto& mapping : _mappings) {
2282
9.46M
        if (mapping.global_index == global_index) {
2283
755k
            return &mapping;
2284
755k
        }
2285
9.46M
    }
2286
166
    return nullptr;
2287
755k
}
2288
2289
152k
ColumnMapping* TableColumnMapper::_find_filter_mapping(GlobalIndex global_index) {
2290
152k
    if (auto* mapping = _find_mapping(global_index); mapping != nullptr) {
2291
152k
        return mapping;
2292
152k
    }
2293
52
    for (auto& mapping : _hidden_mappings) {
2294
4
        if (mapping.global_index == global_index) {
2295
4
            return &mapping;
2296
4
        }
2297
4
    }
2298
48
    return nullptr;
2299
52
}
2300
2301
Status TableColumnMapper::localize_filters(const std::vector<TableFilter>& table_filters,
2302
                                           FileScanRequest* file_request,
2303
122k
                                           RuntimeState* runtime_state) {
2304
122k
    std::set<LocalColumnId> localized_predicate_columns;
2305
122k
    FilterProjectionMap filter_projections;
2306
122k
    auto filter_mappings = _filter_visible_mappings();
2307
122k
    RETURN_IF_ERROR(build_nested_struct_filter_projection_map(table_filters, filter_mappings,
2308
122k
                                                              &filter_projections));
2309
122k
    for (const auto& table_filter : table_filters) {
2310
81.1k
        for (const auto& global_index : table_filter.global_indices) {
2311
81.1k
            auto* mapping = _find_filter_mapping(global_index);
2312
81.1k
            if (mapping == nullptr || !mapping->file_local_id.has_value() ||
2313
81.1k
                !filter_conversion_has_local_source(mapping->filter_conversion)) {
2314
7.23k
                continue;
2315
7.23k
            }
2316
73.9k
            RETURN_IF_ERROR(add_scan_column(file_request, mapping, enable_lazy_materialization(),
2317
73.9k
                                            force_full_complex_scan_projection(),
2318
73.9k
                                            &filter_projections));
2319
73.9k
        }
2320
79.8k
    }
2321
    // Rebuild the file type for every scan-local mapping before expression rewrite. Predicate-only
2322
    // hidden mappings must see the same projected file type as the file reader will produce.
2323
533k
    for (auto& mapping : _mappings) {
2324
533k
        if (mapping.file_local_id.has_value() &&
2325
533k
            file_request->local_positions.contains(LocalColumnId(*mapping.file_local_id))) {
2326
506k
            RETURN_IF_ERROR(apply_scan_projection_to_mapping_file_type(*file_request, &mapping));
2327
506k
        }
2328
533k
    }
2329
122k
    for (auto& mapping : _hidden_mappings) {
2330
2
        if (mapping.file_local_id.has_value() &&
2331
2
            file_request->local_positions.contains(LocalColumnId(*mapping.file_local_id))) {
2332
2
            RETURN_IF_ERROR(apply_scan_projection_to_mapping_file_type(*file_request, &mapping));
2333
2
        }
2334
2
    }
2335
122k
    RETURN_IF_ERROR(_build_filter_entries(*file_request));
2336
2337
    // Build the complete table-slot rewrite map after all predicate columns have been assigned.
2338
    // This keeps expression localization independent from filter iteration order.
2339
122k
    filter_mappings = _filter_visible_mappings();
2340
122k
    const auto global_to_file_slot = build_file_slot_rewrite_map(filter_mappings, _filter_entries);
2341
122k
    for (const auto& table_filter : table_filters) {
2342
79.9k
        if (table_filter.conjunct != nullptr && table_filter.conjunct->root() != nullptr) {
2343
79.9k
            const auto root = table_filter.conjunct->root();
2344
79.9k
            const auto impl = root->get_impl();
2345
79.9k
            const auto predicate = impl != nullptr ? impl : root;
2346
79.9k
            if (!predicate->is_deterministic() ||
2347
79.9k
                !table_filter_has_only_local_entries(table_filter, _filter_entries)) {
2348
6.99k
                continue;
2349
6.99k
            }
2350
            // Scanner evaluates the original conjunct after final materialization. Only predicates
2351
            // whose result is stable across repeated execution may also run as a file-local copy.
2352
72.9k
            RewriteContext rewrite_context {.runtime_state = runtime_state};
2353
72.9k
            VExprSPtr rewrite_root;
2354
72.9k
            Status clone_status;
2355
72.9k
            try {
2356
72.9k
                clone_status = clone_table_expr_tree(table_filter.conjunct->root(), &rewrite_root);
2357
72.9k
            } catch ([[maybe_unused]] const Exception& e) {
2358
                // Some table filters contain complex intermediate values, for example
2359
                // `element_at(MAP_VALUES(m)[1], 'age') > 30`. The current file-local rewrite only
2360
                // understands top-level slots and struct-element paths rooted at top-level slots;
2361
                // cloning such expressions can hit the generic TExpr complex-type limitation.
2362
                // Leave them above TableReader, where Scanner evaluates the original table-level
2363
                // conjunct after final materialization.
2364
0
#ifndef NDEBUG
2365
0
                return Status::InternalError(
2366
0
                        "Failed to clone table filter for file-local rewrite: {}, expr={}",
2367
0
                        e.to_string(), table_filter.conjunct->root()->debug_string());
2368
#else
2369
                continue;
2370
#endif
2371
0
            } catch ([[maybe_unused]] const std::exception& e) {
2372
0
#ifndef NDEBUG
2373
0
                return Status::InternalError(
2374
0
                        "Failed to clone table filter for file-local rewrite: {}, expr={}",
2375
0
                        e.what(), table_filter.conjunct->root()->debug_string());
2376
#else
2377
                continue;
2378
#endif
2379
0
            }
2380
72.9k
            if (!clone_status.ok()) {
2381
0
#ifndef NDEBUG
2382
0
                return Status::InternalError(
2383
0
                        "Failed to clone table filter for file-local rewrite: {}, expr={}",
2384
0
                        clone_status.to_string(), table_filter.conjunct->root()->debug_string());
2385
#else
2386
                continue;
2387
#endif
2388
0
            }
2389
72.9k
            bool can_localize = true;
2390
72.9k
            auto localized_root = rewrite_table_expr_to_file_expr(rewrite_root, global_to_file_slot,
2391
72.9k
                                                                  filter_mappings, &rewrite_context,
2392
72.9k
                                                                  &can_localize);
2393
72.9k
            if (!can_localize) {
2394
2.59k
                continue;
2395
2.59k
            }
2396
70.3k
            auto localized_conjunct = VExprContext::create_shared(std::move(localized_root));
2397
70.3k
            RETURN_IF_ERROR(rewrite_context.prepare_created_exprs(localized_conjunct.get()));
2398
70.3k
            file_request->conjuncts.push_back(std::move(localized_conjunct));
2399
71.0k
            for (const auto global_index : table_filter.global_indices) {
2400
71.0k
                const auto* mapping = _find_filter_mapping(global_index);
2401
71.0k
                if (mapping != nullptr && mapping->file_local_id.has_value() &&
2402
71.0k
                    filter_conversion_has_local_source(mapping->filter_conversion)) {
2403
70.9k
                    localized_predicate_columns.emplace(*mapping->file_local_id);
2404
70.9k
                }
2405
71.0k
            }
2406
70.3k
        }
2407
79.9k
    }
2408
2409
    // Candidate columns are added before expression rewriting because their file-block positions
2410
    // are needed to localize slot refs. If rewriting rejects every filter that references a visible
2411
    // column, move its already-merged output/filter projection to the lazy non-predicate set
2412
    // instead of forcing it through the eager predicate path.
2413
533k
    for (auto& mapping : _mappings) {
2414
533k
        if (!mapping.file_local_id.has_value()) {
2415
26.6k
            continue;
2416
26.6k
        }
2417
506k
        const auto local_id = LocalColumnId(*mapping.file_local_id);
2418
506k
        if (localized_predicate_columns.contains(local_id)) {
2419
61.2k
            continue;
2420
61.2k
        }
2421
445k
        const auto predicate_it = std::ranges::find_if(
2422
445k
                file_request->predicate_columns, [local_id](const LocalColumnIndex& projection) {
2423
59.1k
                    return projection.column_id() == local_id;
2424
59.1k
                });
2425
445k
        if (predicate_it == file_request->predicate_columns.end()) {
2426
442k
            continue;
2427
442k
        }
2428
2.80k
        file_request->non_predicate_columns.push_back(std::move(*predicate_it));
2429
2.80k
        file_request->predicate_columns.erase(predicate_it);
2430
2.80k
    }
2431
122k
    return Status::OK();
2432
122k
}
2433
2434
const ColumnDefinition* TableColumnMapper::_find_file_field(
2435
        const ColumnDefinition& table_column,
2436
485k
        const std::vector<ColumnDefinition>& file_schema) const {
2437
485k
    if (table_column.name.starts_with(BeConsts::GLOBAL_ROWID_COL)) {
2438
258k
        const auto field_it = std::ranges::find_if(file_schema, [](const ColumnDefinition& field) {
2439
258k
            return field.column_type == ColumnType::GLOBAL_ROWID;
2440
258k
        });
2441
6.26k
        return field_it == file_schema.end() ? nullptr : &*field_it;
2442
6.26k
    }
2443
479k
    if (_options.mode == TableColumnMappingMode::BY_FIELD_ID) {
2444
227k
        return find_column_by_field_id(table_column, file_schema,
2445
227k
                                       _options.allow_idless_complex_wrapper_projection);
2446
227k
    }
2447
252k
    return matcher_for_mode(_options.mode).find(table_column, file_schema);
2448
479k
}
2449
2450
Status TableColumnMapper::_create_direct_mapping(const ColumnDefinition& table_column,
2451
                                                 const ColumnDefinition& file_field,
2452
832k
                                                 ColumnMapping* mapping) const {
2453
832k
    DORIS_CHECK(mapping != nullptr);
2454
832k
    DORIS_CHECK(file_field.local_id >= 0 || file_field.local_id == GLOBAL_ROWID_COLUMN_ID);
2455
832k
    mapping->file_local_id = file_field.local_id;
2456
832k
    mapping->table_column_name = table_column.name;
2457
832k
    mapping->file_column_name = file_field.name;
2458
832k
    mapping->original_file_type = file_field.type;
2459
832k
    mapping->original_file_children = file_field.children;
2460
832k
    mapping->projected_file_children = file_field.children;
2461
832k
    mapping->timestamp_is_adjusted_to_utc = file_field.timestamp_is_adjusted_to_utc;
2462
832k
    mapping->file_type = file_field.type;
2463
832k
    mapping->is_trivial = mapping_can_use_file_column_directly(*mapping);
2464
832k
    mapping->filter_conversion = direct_filter_conversion(*mapping);
2465
832k
    mapping->child_mappings.clear();
2466
2467
832k
    auto [table_children, synthesized_table_children] =
2468
832k
            prepare_table_children_for_mapping(table_column, mapping->file_type);
2469
2470
832k
    if (!table_children.empty()) {
2471
210k
        if (!is_complex_type(remove_nullable(mapping->file_type)->get_primitive_type())) {
2472
0
            return Status::NotSupported(
2473
0
                    "Cannot map complex table column '{}' to scalar parquet column '{}', table "
2474
0
                    "type={}, file type={}",
2475
0
                    table_column.name, file_field.name, mapping->table_type->get_name(),
2476
0
                    mapping->file_type->get_name());
2477
0
        }
2478
210k
        RETURN_IF_ERROR(validate_file_schema_children(file_field));
2479
210k
        std::vector<int32_t> synthesized_used_file_child_ids;
2480
542k
        for (size_t table_child_idx = 0; table_child_idx < table_children.size();
2481
332k
             ++table_child_idx) {
2482
332k
            const auto& table_child = table_children[table_child_idx];
2483
332k
            const auto* file_child =
2484
332k
                    find_file_child_for_mapping(table_child, file_field, _options.mode,
2485
332k
                                                table_child_idx, synthesized_table_children);
2486
332k
            if (file_child == nullptr && !synthesized_table_children &&
2487
332k
                _options.mode == TableColumnMappingMode::BY_FIELD_ID &&
2488
332k
                _options.allow_idless_complex_wrapper_projection) {
2489
                // Parquet can retain an ID-less wrapper at any depth when a selected descendant
2490
                // has an ID; apply the same opt-in fallback used for root lookup recursively.
2491
2.44k
                file_child = _find_file_field(table_child, file_field.children);
2492
2.44k
            }
2493
332k
            if (synthesized_table_children && file_child != nullptr) {
2494
10
                const auto file_child_id = file_child->file_local_id();
2495
10
                if (std::ranges::find(synthesized_used_file_child_ids, file_child_id) !=
2496
10
                    synthesized_used_file_child_ids.end()) {
2497
2
                    file_child = nullptr;
2498
2
                    for (const auto& candidate : file_field.children) {
2499
2
                        const auto candidate_id = candidate.file_local_id();
2500
2
                        if (std::ranges::find(synthesized_used_file_child_ids, candidate_id) ==
2501
2
                            synthesized_used_file_child_ids.end()) {
2502
2
                            file_child = &candidate;
2503
2
                            break;
2504
2
                        }
2505
2
                    }
2506
2
                }
2507
10
                if (file_child != nullptr) {
2508
10
                    synthesized_used_file_child_ids.push_back(file_child->file_local_id());
2509
10
                }
2510
10
            }
2511
332k
            if (file_child == nullptr) {
2512
5.59k
                if (table_child.default_expr == nullptr &&
2513
5.59k
                    table_child.initial_default_value.has_value()) {
2514
0
                    return Status::InvalidArgument(
2515
0
                            "Missing typed initial-default expression for table field '{}'",
2516
0
                            table_child.name);
2517
0
                }
2518
5.59k
                if (_options.reject_missing_required_field && table_child.is_optional.has_value() &&
2519
5.59k
                    !*table_child.is_optional && table_child.default_expr == nullptr) {
2520
1
                    return Status::InvalidArgument("Missing required field: {}", table_child.name);
2521
1
                }
2522
5.59k
                ColumnMapping child_mapping;
2523
5.59k
                child_mapping.table_column_name = table_child.name;
2524
5.59k
                child_mapping.file_column_name = table_child.name;
2525
5.59k
                child_mapping.table_type = table_child.type;
2526
5.59k
                child_mapping.file_type = table_child.type;
2527
5.59k
                child_mapping.default_expr = table_child.default_expr;
2528
5.59k
                child_mapping.filter_conversion = FilterConversionType::FINALIZE_ONLY;
2529
5.59k
                mapping->child_mappings.push_back(std::move(child_mapping));
2530
5.59k
                continue;
2531
5.59k
            }
2532
326k
            ColumnMapping child_mapping;
2533
326k
            child_mapping.table_column_name = table_child.name;
2534
326k
            child_mapping.table_type = table_child.type;
2535
326k
            RETURN_IF_ERROR(_create_direct_mapping(table_child, *file_child, &child_mapping));
2536
326k
            mapping->child_mappings.push_back(std::move(child_mapping));
2537
326k
        }
2538
210k
        if (needs_projected_file_type_rebuild(*mapping)) {
2539
            // If complex projection prunes some children, we have to rebuild the projected file type to make sure the reader expression can find the correct child types by name.
2540
19.7k
            RETURN_IF_ERROR(rebuild_projected_file_children_and_type(
2541
19.7k
                    mapping->file_type, mapping->original_file_children, mapping->child_mappings,
2542
19.7k
                    &mapping->projected_file_children, &mapping->file_type));
2543
19.7k
            DCHECK(mapping->table_type != nullptr);
2544
19.7k
            mapping->is_trivial = mapping_can_use_file_column_directly(*mapping);
2545
19.7k
            mapping->filter_conversion = projected_filter_conversion(*mapping);
2546
19.7k
        }
2547
210k
    }
2548
832k
    return Status::OK();
2549
832k
}
2550
2551
} // namespace doris::format