[refactor](be) Unify storage reader column ordinals

be/src/cloud/cloud_schema_change_job.cpp

@@ -38,6 +38,7 @@
 #include "storage/rowset/beta_rowset.h"
 #include "storage/rowset/rowset.h"
 #include "storage/rowset/rowset_factory.h"
+#include "storage/schema.h"
 #include "storage/storage_engine.h"
 #include "storage/tablet/tablet.h"
 #include "storage/tablet/tablet_fwd.h"
@@ -237,48 +238,48 @@ Status CloudSchemaChangeJob::process_alter_tablet(const TAlterTabletReqV2& reque
 
     // FIXME(cyx): Should trigger compaction on base_tablet if there are too many rowsets to convert.
 
-    // Create a new tablet schema, should merge with dropped columns in light weight schema change
+    // Build the visible base schema. Historical delete columns are resolved by DeleteHandler.
     _base_tablet_schema = std::make_shared<TabletSchema>();
     _base_tablet_schema->update_tablet_columns(*_base_tablet->tablet_schema(), request.columns);
     _new_tablet_schema = _new_tablet->tablet_schema();
 
-    std::vector<ColumnId> return_columns;
-    return_columns.resize(_base_tablet_schema->num_columns());
-    std::iota(return_columns.begin(), return_columns.end(), 0);
+    ReadSchemaSPtr read_schema = std::make_shared<ReadSchema>(_base_tablet_schema->columns());
 
     // delete handlers to filter out deleted rows
     DeleteHandler delete_handler;
     std::vector<RowsetMetaSharedPtr> delete_predicates;
     for (auto& split : rs_splits) {
         auto& rs_meta = split.rs_reader->rowset()->rowset_meta();
         if (rs_meta->has_delete_predicate()) {
-            _base_tablet_schema->merge_dropped_columns(*rs_meta->tablet_schema());
             delete_predicates.push_back(rs_meta);
         }
     }
-    RETURN_IF_ERROR(delete_handler.init(_base_tablet_schema, delete_predicates,
-                                        start_resp.alter_version()));
+    std::vector<TabletColumn> dropped_columns;
+    RETURN_IF_ERROR(delete_handler.init(delete_predicates, start_resp.alter_version(), read_schema,
+                                        dropped_columns));
+    for (auto& column : dropped_columns) {
+        read_schema->append_column(std::make_shared<TabletColumn>(std::move(column)));
+    }
 
     // reader_context is stack variables, it's lifetime MUST keep the same with rs_readers
     RowsetReaderContext reader_context;
     reader_context.reader_type = ReaderType::READER_ALTER_TABLE;
     reader_context.tablet_schema = _base_tablet_schema;
     reader_context.need_ordered_result = true;
     reader_context.delete_handler = &delete_handler;
-    reader_context.return_columns = &return_columns;
-    reader_context.sequence_id_idx = reader_context.tablet_schema->sequence_col_idx();
+    reader_context.read_schema = read_schema;
     reader_context.is_unique = _base_tablet->keys_type() == UNIQUE_KEYS;
     reader_context.batch_size = ALTER_TABLE_BATCH_SIZE;
     reader_context.delete_bitmap = _base_tablet->tablet_meta()->delete_bitmap_ptr();
     reader_context.version = Version(0, start_resp.alter_version());
     std::vector<uint32_t> cluster_key_idxes;
     if (!_base_tablet_schema->cluster_key_uids().empty()) {
         for (const auto& uid : _base_tablet_schema->cluster_key_uids()) {
-            cluster_key_idxes.emplace_back(_base_tablet_schema->field_index(uid));
+            cluster_key_idxes.emplace_back(read_schema->ordinal_by_uid(uid));
         }
         reader_context.read_orderby_key_columns = &cluster_key_idxes;
         reader_context.is_unique = false;
-        reader_context.sequence_id_idx = -1;
+        reader_context.use_sequence_column_for_merge_order = false;
     }
 
     for (auto& split : rs_splits) {

be/src/exec/operator/file_scan_operator.h

@@ -112,17 +112,6 @@ class FileScanOperatorX final : public ScanOperatorX<FileScanLocalState> {
         return _batch_split_mode ? 1 : ScanOperatorX<FileScanLocalState>::parallelism(state);
     }
 
-    int get_column_id(const std::string& col_name) const override {
-        int column_id_counter = 0;
-        for (const auto& slot : _output_tuple_desc->slots()) {
-            if (slot->col_name() == col_name) {
-                return column_id_counter;
-            }
-            column_id_counter++;
-        }
-        return column_id_counter;
-    }
-
     bool can_push_down_column_predicate(const SlotDescriptor* slot) const override;
 
 private:

be/src/exec/operator/olap_scan_operator.cpp

@@ -298,8 +298,6 @@ Status OlapScanLocalState::_init_profile() {
             ADD_CHILD_TIMER(_scanner_profile, "TabletReaderInitTimer", "ReaderInitTime");
     _tablet_reader_capture_rs_readers_timer = ADD_CHILD_TIMER(
             _scanner_profile, "TabletReaderCaptureRsReadersTimer", "TabletReaderInitTimer");
-    _tablet_reader_init_return_columns_timer = ADD_CHILD_TIMER(
-            _scanner_profile, "TabletReaderInitReturnColumnsTimer", "TabletReaderInitTimer");
     _tablet_reader_init_keys_param_timer = ADD_CHILD_TIMER(
             _scanner_profile, "TabletReaderInitKeysParamTimer", "TabletReaderInitTimer");
     _tablet_reader_init_orderby_keys_param_timer = ADD_CHILD_TIMER(
@@ -329,8 +327,8 @@ Status OlapScanLocalState::_init_profile() {
 
     _segment_iterator_init_timer =
             ADD_CHILD_TIMER(_scanner_profile, "SegmentIteratorInitTimer", "BlockFetchTime");
-    _segment_iterator_init_return_column_iterators_timer =
-            ADD_CHILD_TIMER(_scanner_profile, "SegmentIteratorInitReturnColumnIteratorsTimer",
+    _segment_iterator_init_column_iterators_timer =
+            ADD_CHILD_TIMER(_scanner_profile, "SegmentIteratorInitColumnIteratorsTimer",
                             "SegmentIteratorInitTimer");
     _segment_iterator_init_index_iterators_timer = ADD_CHILD_TIMER(
             _scanner_profile, "SegmentIteratorInitIndexIteratorsTimer", "SegmentIteratorInitTimer");

be/src/exec/operator/olap_scan_operator.h

@@ -301,7 +301,6 @@ class OlapScanLocalState final : public ScanLocalState<OlapScanLocalState> {
     // timer about tablet reader
     RuntimeProfile::Counter* _tablet_reader_init_timer = nullptr;
     RuntimeProfile::Counter* _tablet_reader_capture_rs_readers_timer = nullptr;
-    RuntimeProfile::Counter* _tablet_reader_init_return_columns_timer = nullptr;
     RuntimeProfile::Counter* _tablet_reader_init_keys_param_timer = nullptr;
     RuntimeProfile::Counter* _tablet_reader_init_orderby_keys_param_timer = nullptr;
     RuntimeProfile::Counter* _tablet_reader_init_conditions_param_timer = nullptr;
@@ -318,7 +317,7 @@ class OlapScanLocalState final : public ScanLocalState<OlapScanLocalState> {
     RuntimeProfile::Counter* _rowset_reader_load_segments_timer = nullptr;
 
     RuntimeProfile::Counter* _segment_iterator_init_timer = nullptr;
-    RuntimeProfile::Counter* _segment_iterator_init_return_column_iterators_timer = nullptr;
+    RuntimeProfile::Counter* _segment_iterator_init_column_iterators_timer = nullptr;
     RuntimeProfile::Counter* _segment_iterator_init_index_iterators_timer = nullptr;
     RuntimeProfile::Counter* _segment_iterator_init_segment_prefetchers_timer = nullptr;
 
@@ -375,14 +374,6 @@ class OlapScanOperatorX final : public ScanOperatorX<OlapScanLocalState> {
 
     Status prepare(RuntimeState* state) override;
 
-    int get_column_id(const std::string& col_name) const override {
-        if (!_tablet_schema) {
-            return -1;
-        }
-        const auto& column = *DORIS_TRY(_tablet_schema->column(col_name));
-        return _tablet_schema->field_index(column.unique_id());
-    }
-
 private:
     friend class OlapScanLocalState;
     TOlapScanNode _olap_scan_node;

be/src/exec/operator/scan_operator.cpp

@@ -51,6 +51,7 @@
 #include "runtime/descriptors.h"
 #include "runtime/runtime_profile.h"
 #include "runtime/runtime_profile_counter_names.h"
+#include "storage/predicate/like_column_predicate.h"
 #include "storage/predicate/null_predicate.h"
 #include "storage/predicate/predicate_creator.h"
 
@@ -482,8 +483,10 @@ Status ScanLocalState<Derived>::_normalize_predicate(VExprContext* context, cons
                     }
                     // `node_type` of function filter is FUNCTION_CALL or COMPOUND_PRED
                     if (state()->enable_function_pushdown()) {
-                        RETURN_IF_PUSH_DOWN(_normalize_function_filters(context, slot, &pdt),
-                                            status);
+                        RETURN_IF_PUSH_DOWN(
+                                _normalize_function_filters(
+                                        context, slot, _slot_id_to_predicates[slot->id()], &pdt),
+                                status);
                     }
                 },
                 *range);
@@ -529,7 +532,7 @@ Status ScanLocalStateBase::_normalize_bloom_filter(
     *pdt = _should_push_down_bloom_filter();
     if (*pdt != PushDownType::UNACCEPTABLE) {
         pred = create_bloom_filter_predicate(
-                _parent->intermediate_row_desc().get_column_id(slot->id()), slot->col_name(),
+                output_tuple_desc()->get_column_id(slot->id()), slot->col_name(),
                 slot->type()->get_primitive_type() == TYPE_VARIANT ? expr->get_child(0)->data_type()
                                                                    : slot->type(),
                 expr->get_bloom_filter_func());
@@ -562,8 +565,9 @@ Status ScanLocalStateBase::_normalize_topn_filter(
     return Status::OK();
 }
 
-Status ScanLocalStateBase::_normalize_function_filters(VExprContext* expr_ctx, SlotDescriptor* slot,
-                                                       PushDownType* pdt) {
+Status ScanLocalStateBase::_normalize_function_filters(
+        VExprContext* expr_ctx, SlotDescriptor* slot,
+        std::vector<std::shared_ptr<ColumnPredicate>>& predicates, PushDownType* pdt) {
     auto expr = expr_ctx->root()->is_rf_wrapper() ? expr_ctx->root()->get_impl() : expr_ctx->root();
     bool opposite = false;
     VExpr* fn_expr = expr.get();
@@ -580,8 +584,10 @@ Status ScanLocalStateBase::_normalize_function_filters(VExprContext* expr_ctx, S
         RETURN_IF_ERROR(_should_push_down_function_filter(assert_cast<VectorizedFnCall*>(fn_expr),
                                                           expr_ctx, &val, &fn_ctx, temp_pdt));
         if (temp_pdt != PushDownType::UNACCEPTABLE) {
-            std::string col = slot->col_name();
-            _push_down_functions.emplace_back(opposite, col, fn_ctx, val);
+            const auto column_id =
+                    cast_set<uint32_t>(output_tuple_desc()->get_column_id(slot->id()));
+            predicates.emplace_back(LikeColumnPredicate::create_shared(
+                    opposite, column_id, slot->col_name(), fn_ctx, val));
             *pdt = temp_pdt;
         }
     }
@@ -786,15 +792,13 @@ Status ScanLocalStateBase::_normalize_in_predicate(
         }
     }
     pred = is_in ? create_in_list_predicate<PredicateType::IN_LIST>(
-                           _parent->intermediate_row_desc().get_column_id(slot->id()),
-                           slot->col_name(),
+                           output_tuple_desc()->get_column_id(slot->id()), slot->col_name(),
                            slot->type()->get_primitive_type() == TYPE_VARIANT
                                    ? root->get_child(0)->data_type()
                                    : slot->type(),
                            hybrid_set, false)
                  : create_in_list_predicate<PredicateType::NOT_IN_LIST>(
-                           _parent->intermediate_row_desc().get_column_id(slot->id()),
-                           slot->col_name(),
+                           output_tuple_desc()->get_column_id(slot->id()), slot->col_name(),
                            slot->type()->get_primitive_type() == TYPE_VARIANT
                                    ? root->get_child(0)->data_type()
                                    : slot->type(),
@@ -843,47 +847,47 @@ Status ScanLocalStateBase::_normalize_binary_predicate(
         switch (op) {
         case SQLFilterOp::FILTER_EQ:
             pred = create_comparison_predicate<PredicateType::EQ>(
-                    _parent->intermediate_row_desc().get_column_id(slot->id()), slot->col_name(),
+                    output_tuple_desc()->get_column_id(slot->id()), slot->col_name(),
                     slot->type()->get_primitive_type() == TYPE_VARIANT
                             ? root->get_child(0)->data_type()
                             : slot->type(),
                     value, false);
             break;
         case SQLFilterOp::FILTER_NE:
             pred = create_comparison_predicate<PredicateType::NE>(
-                    _parent->intermediate_row_desc().get_column_id(slot->id()), slot->col_name(),
+                    output_tuple_desc()->get_column_id(slot->id()), slot->col_name(),
                     slot->type()->get_primitive_type() == TYPE_VARIANT
                             ? root->get_child(0)->data_type()
                             : slot->type(),
                     value, false);
             break;
         case SQLFilterOp::FILTER_LESS:
             pred = create_comparison_predicate<PredicateType::LT>(
-                    _parent->intermediate_row_desc().get_column_id(slot->id()), slot->col_name(),
+                    output_tuple_desc()->get_column_id(slot->id()), slot->col_name(),
                     slot->type()->get_primitive_type() == TYPE_VARIANT
                             ? root->get_child(0)->data_type()
                             : slot->type(),
                     value, false);
             break;
         case SQLFilterOp::FILTER_LARGER:
             pred = create_comparison_predicate<PredicateType::GT>(
-                    _parent->intermediate_row_desc().get_column_id(slot->id()), slot->col_name(),
+                    output_tuple_desc()->get_column_id(slot->id()), slot->col_name(),
                     slot->type()->get_primitive_type() == TYPE_VARIANT
                             ? root->get_child(0)->data_type()
                             : slot->type(),
                     value, false);
             break;
         case SQLFilterOp::FILTER_LESS_OR_EQUAL:
             pred = create_comparison_predicate<PredicateType::LE>(
-                    _parent->intermediate_row_desc().get_column_id(slot->id()), slot->col_name(),
+                    output_tuple_desc()->get_column_id(slot->id()), slot->col_name(),
                     slot->type()->get_primitive_type() == TYPE_VARIANT
                             ? root->get_child(0)->data_type()
                             : slot->type(),
                     value, false);
             break;
         case SQLFilterOp::FILTER_LARGER_OR_EQUAL:
             pred = create_comparison_predicate<PredicateType::GE>(
-                    _parent->intermediate_row_desc().get_column_id(slot->id()), slot->col_name(),
+                    output_tuple_desc()->get_column_id(slot->id()), slot->col_name(),
                     slot->type()->get_primitive_type() == TYPE_VARIANT
                             ? root->get_child(0)->data_type()
                             : slot->type(),
@@ -978,17 +982,15 @@ Status ScanLocalStateBase::_normalize_is_null_predicate(
 
     auto fn_call = assert_cast<VectorizedFnCall*>(root.get());
     if (fn_call->fn().name.function_name == "is_null_pred") {
-        pred = NullPredicate::create_shared(
-                _parent->intermediate_row_desc().get_column_id(slot->id()), slot->col_name(), true,
-                T);
+        pred = NullPredicate::create_shared(output_tuple_desc()->get_column_id(slot->id()),
+                                            slot->col_name(), true, T);
         auto temp_range = ColumnValueRange<T>::create_empty_column_value_range(
                 slot->is_nullable(), range.precision(), range.scale());
         temp_range.set_contain_null(true);
         range.intersection(temp_range);
     } else if (fn_call->fn().name.function_name == "is_not_null_pred") {
-        pred = NullPredicate::create_shared(
-                _parent->intermediate_row_desc().get_column_id(slot->id()), slot->col_name(), false,
-                T);
+        pred = NullPredicate::create_shared(output_tuple_desc()->get_column_id(slot->id()),
+                                            slot->col_name(), false, T);
         auto temp_range = ColumnValueRange<T>::create_empty_column_value_range(
                 slot->is_nullable(), range.precision(), range.scale());
         temp_range.set_contain_null(false);
@@ -1284,8 +1286,7 @@ Status ScanOperatorX<LocalStateType>::prepare(RuntimeState* state) {
                                                    .slot_ref.slot_id];
             DCHECK(s != nullptr);
             if (can_push_down_column_predicate(s)) {
-                auto col_name = s->col_name();
-                cid = get_column_id(col_name);
+                cid = _output_tuple_desc->get_column_id(s->id());
             }
         }
         RETURN_IF_ERROR(state->get_query_ctx()->get_runtime_predicate(id).init_target(

be/src/exec/operator/scan_operator.h

@@ -32,7 +32,6 @@
 #include "exec/runtime_filter/runtime_filter_partition_pruner.h"
 #include "exec/scan/scan_node.h"
 #include "exec/scan/scanner_context.h"
-#include "exprs/function_filter.h"
 #include "exprs/vectorized_fn_call.h"
 #include "exprs/vin_predicate.h"
 #include "runtime/descriptors.h"
@@ -156,9 +155,6 @@ class ScanLocalStateBase : public PipelineXLocalState<> {
     // Moved from ScanLocalState<Derived> to avoid re-instantiation for each Derived type.
     std::atomic<bool> _eos = false;
     int _max_pushdown_conditions_per_column = 1024;
-    // Save all function predicates which may be pushed down to data source.
-    std::vector<FunctionFilter> _push_down_functions;
-
     // Virtual methods with default implementations; overridden by subclasses when supported.
     // Declared here so that the normalize methods below (non-Derived-template) can call them.
     virtual bool _push_down_topn(const RuntimePredicate& predicate) { return false; }
@@ -199,6 +195,7 @@ class ScanLocalStateBase : public PipelineXLocalState<> {
                                   std::vector<std::shared_ptr<ColumnPredicate>>& predicates,
                                   PushDownType* pdt);
     Status _normalize_function_filters(VExprContext* expr_ctx, SlotDescriptor* slot,
+                                       std::vector<std::shared_ptr<ColumnPredicate>>& predicates,
                                        PushDownType* pdt);
 
     // Inner PrimitiveType-template methods. Moved to base to avoid N(Derived)×M(PrimitiveType)
@@ -345,7 +342,6 @@ class ScanLocalState : public ScanLocalStateBase {
     // Parsed from conjuncts
     phmap::flat_hash_map<int, ColumnValueRangeType> _slot_id_to_value_range;
     phmap::flat_hash_map<int, std::vector<std::shared_ptr<ColumnPredicate>>> _slot_id_to_predicates;
-    std::vector<std::shared_ptr<MutilColumnBlockPredicate>> _or_predicates;
 
     std::vector<std::shared_ptr<Dependency>> _filter_dependencies;
 
@@ -386,8 +382,6 @@ class ScanOperatorX : public OperatorX<LocalStateType> {
         return &_parsed_partition_boundaries;
     }
 
-    [[nodiscard]] virtual int get_column_id(const std::string& col_name) const { return -1; }
-
     [[nodiscard]] virtual bool can_push_down_column_predicate(const SlotDescriptor*) const {
         return true;
     }