[Enhancement] compute unused column by be (backport #60462) (#65844)

mergify[bot] · Seaven · web-flow · commit b8bd3068bf93 · 2025-11-21T08:16:56.000Z
Signed-off-by: Seaven &lt;seaven_7@qq.com&gt;
Co-authored-by: Seaven &lt;seaven_7@qq.com&gt;
diff --git a/be/src/connector/lake_connector.cpp b/be/src/connector/lake_connector.cpp
@@ -216,7 +216,8 @@ Status LakeDataSource::init_unused_output_columns(const std::vector<std::string>
     return Status::OK();
 }
 
-Status LakeDataSource::init_scanner_columns(std::vector<uint32_t>& scanner_columns) {
+Status LakeDataSource::init_scanner_columns(std::vector<uint32_t>& scanner_columns,
+                                            std::vector<uint32_t>& reader_columns) {
     for (auto slot : *_slots) {
         DCHECK(slot->is_materialized());
         int32_t index = _tablet_schema->field_index(slot->col_name());
@@ -237,6 +238,23 @@ Status LakeDataSource::init_scanner_columns(std::vector<uint32_t>& scanner_colum
     if (scanner_columns.empty()) {
         return Status::InternalError("failed to build storage scanner, no materialized slot!");
     }
+
+    // Return columns
+    if (_params.skip_aggregation) {
+        reader_columns = scanner_columns;
+    } else {
+        for (size_t i = 0; i < _tablet_schema->num_key_columns(); i++) {
+            reader_columns.push_back(i);
+        }
+        for (auto index : scanner_columns) {
+            if (!_tablet_schema->column(index).is_key()) {
+                reader_columns.push_back(index);
+            }
+        }
+    }
+    // Actually only the key columns need to be sorted by id, here we check all
+    // for simplicity.
+    DCHECK(std::is_sorted(reader_columns.begin(), reader_columns.end()));
     return Status::OK();
 }
 
@@ -249,9 +267,7 @@ void LakeDataSource::decide_chunk_size(bool has_predicate) {
     }
 }
 
-Status LakeDataSource::init_reader_params(const std::vector<OlapScanRange*>& key_ranges,
-                                          const std::vector<uint32_t>& scanner_columns,
-                                          std::vector<uint32_t>& reader_columns) {
+Status LakeDataSource::init_reader_params(const std::vector<OlapScanRange*>& key_ranges) {
     const TLakeScanNode& thrift_lake_scan_node = _provider->_t_lake_scan_node;
     bool skip_aggregation = thrift_lake_scan_node.is_preaggregation;
     auto parser = _obj_pool.add(new OlapPredicateParser(_tablet_schema));
@@ -282,6 +298,23 @@ Status LakeDataSource::init_reader_params(const std::vector<OlapScanRange*>& key
     _params.pred_tree = PredicateTree::create(std::move(pushdown_pred_root));
     _non_pushdown_pred_tree = PredicateTree::create(std::move(non_pushdown_pred_root));
 
+    for (const auto& cid : _non_pushdown_pred_tree.column_ids()) {
+        _unused_output_column_ids.erase(cid);
+    }
+
+    std::vector<ExprContext*> not_pushdown_conjuncts;
+    _conjuncts_manager->get_not_push_down_conjuncts(&not_pushdown_conjuncts);
+    std::unordered_set<SlotId> conjuncts_slot_ids;
+    for (auto* expr : not_pushdown_conjuncts) {
+        expr->root()->for_each_slot_id([&conjuncts_slot_ids](SlotId id) { conjuncts_slot_ids.insert(id); });
+    }
+    for (auto& slot : *_slots) {
+        if (conjuncts_slot_ids.contains(slot->id())) {
+            int32_t fid = _tablet_schema->field_index(slot->col_name());
+            _unused_output_column_ids.erase(fid);
+        }
+    }
+
     {
         GlobalDictPredicatesRewriter not_pushdown_predicate_rewriter(*_params.global_dictmaps);
         RETURN_IF_ERROR(not_pushdown_predicate_rewriter.rewrite_predicate(&_obj_pool, _non_pushdown_pred_tree));
@@ -302,22 +335,6 @@ Status LakeDataSource::init_reader_params(const std::vector<OlapScanRange*>& key
         _params.end_key.push_back(key_range->end_scan_range);
     }
 
-    // Return columns
-    if (skip_aggregation) {
-        reader_columns = scanner_columns;
-    } else {
-        for (size_t i = 0; i < _tablet_schema->num_key_columns(); i++) {
-            reader_columns.push_back(i);
-        }
-        for (auto index : scanner_columns) {
-            if (!_tablet_schema->column(index).is_key()) {
-                reader_columns.push_back(index);
-            }
-        }
-    }
-    // Actually only the key columns need to be sorted by id, here we check all
-    // for simplicity.
-    DCHECK(std::is_sorted(reader_columns.begin(), reader_columns.end()));
     return Status::OK();
 }
 
@@ -331,8 +348,8 @@ Status LakeDataSource::init_tablet_reader(RuntimeState* runtime_state) {
     RETURN_IF_ERROR(get_tablet(_scan_range));
     RETURN_IF_ERROR(init_global_dicts(&_params));
     RETURN_IF_ERROR(init_unused_output_columns(thrift_lake_scan_node.unused_output_column_name));
-    RETURN_IF_ERROR(init_scanner_columns(scanner_columns));
-    RETURN_IF_ERROR(init_reader_params(_scanner_ranges, scanner_columns, reader_columns));
+    RETURN_IF_ERROR(init_scanner_columns(scanner_columns, reader_columns));
+    RETURN_IF_ERROR(init_reader_params(_scanner_ranges));
 
     if (_split_context != nullptr) {
         auto split_context = down_cast<const pipeline::LakeSplitContext*>(_split_context);
diff --git a/be/src/connector/lake_connector.h b/be/src/connector/lake_connector.h
@@ -59,10 +59,9 @@ class LakeDataSource final : public DataSource {
     Status get_tablet(const TInternalScanRange& scan_range);
     Status init_global_dicts(TabletReaderParams* params);
     Status init_unused_output_columns(const std::vector<std::string>& unused_output_columns);
-    Status init_scanner_columns(std::vector<uint32_t>& scanner_columns);
+    Status init_scanner_columns(std::vector<uint32_t>& scanner_columns, std::vector<uint32_t>& reader_columns);
     void decide_chunk_size(bool has_predicate);
-    Status init_reader_params(const std::vector<OlapScanRange*>& key_ranges,
-                              const std::vector<uint32_t>& scanner_columns, std::vector<uint32_t>& reader_columns);
+    Status init_reader_params(const std::vector<OlapScanRange*>& key_ranges);
     Status init_tablet_reader(RuntimeState* state);
     Status build_scan_range(RuntimeState* state);
     void init_counter(RuntimeState* state);
diff --git a/be/src/exec/pipeline/scan/olap_chunk_source.cpp b/be/src/exec/pipeline/scan/olap_chunk_source.cpp
@@ -228,9 +228,7 @@ void OlapChunkSource::_decide_chunk_size(bool has_predicate) {
     }
 }
 
-Status OlapChunkSource::_init_reader_params(const std::vector<std::unique_ptr<OlapScanRange>>& key_ranges,
-                                            const std::vector<uint32_t>& scanner_columns,
-                                            std::vector<uint32_t>& reader_columns) {
+Status OlapChunkSource::_init_reader_params(const std::vector<std::unique_ptr<OlapScanRange>>& key_ranges) {
     const TOlapScanNode& thrift_olap_scan_node = _scan_node->thrift_olap_scan_node();
     bool skip_aggregation = thrift_olap_scan_node.is_preaggregation;
     auto parser = _obj_pool.add(new OlapPredicateParser(_tablet_schema));
@@ -289,6 +287,23 @@ Status OlapChunkSource::_init_reader_params(const std::vector<std::unique_ptr<Ol
     _params.pred_tree = PredicateTree::create(std::move(pushdown_pred_root));
     _non_pushdown_pred_tree = PredicateTree::create(std::move(non_pushdown_pred_root));
 
+    for (auto& id : _non_pushdown_pred_tree.column_ids()) {
+        _unused_output_column_ids.erase(id);
+    }
+
+    std::vector<ExprContext*> not_pushdown_conjuncts;
+    _scan_ctx->conjuncts_manager().get_not_push_down_conjuncts(&not_pushdown_conjuncts);
+    std::unordered_set<SlotId> conjuncts_slot_ids;
+    for (auto* expr : not_pushdown_conjuncts) {
+        expr->root()->for_each_slot_id([&conjuncts_slot_ids](SlotId id) { conjuncts_slot_ids.insert(id); });
+    }
+    for (auto& slot : *_slots) {
+        if (conjuncts_slot_ids.contains(slot->id())) {
+            int32_t fid = _tablet_schema->field_index(slot->col_name());
+            _unused_output_column_ids.erase(fid);
+        }
+    }
+
     {
         GlobalDictPredicatesRewriter not_pushdown_predicate_rewriter(*_params.global_dictmaps);
         RETURN_IF_ERROR(not_pushdown_predicate_rewriter.rewrite_predicate(&_obj_pool, _non_pushdown_pred_tree));
@@ -309,26 +324,11 @@ Status OlapChunkSource::_init_reader_params(const std::vector<std::unique_ptr<Ol
         _params.end_key.push_back(key_range->end_scan_range);
     }
 
-    // Return columns
-    if (skip_aggregation) {
-        reader_columns = scanner_columns;
-    } else {
-        for (size_t i = 0; i < _tablet_schema->num_key_columns(); i++) {
-            reader_columns.push_back(i);
-        }
-        for (auto index : scanner_columns) {
-            if (!_tablet_schema->column(index).is_key()) {
-                reader_columns.push_back(index);
-            }
-        }
-    }
-    // Actually only the key columns need to be sorted by id, here we check all
-    // for simplicity.
-    DCHECK(std::is_sorted(reader_columns.begin(), reader_columns.end()));
     return Status::OK();
 }
 
-Status OlapChunkSource::_init_scanner_columns(std::vector<uint32_t>& scanner_columns) {
+Status OlapChunkSource::_init_scanner_columns(std::vector<uint32_t>& scanner_columns,
+                                              std::vector<uint32_t>& reader_columns) {
     for (auto slot : *_slots) {
         DCHECK(slot->is_materialized());
         int32_t index;
@@ -357,6 +357,23 @@ Status OlapChunkSource::_init_scanner_columns(std::vector<uint32_t>& scanner_col
         return Status::InternalError("failed to build storage scanner, no materialized slot!");
     }
 
+    // Return columns
+    if (_params.skip_aggregation) {
+        reader_columns = scanner_columns;
+    } else {
+        for (size_t i = 0; i < _tablet_schema->num_key_columns(); i++) {
+            reader_columns.push_back(i);
+        }
+        for (auto index : scanner_columns) {
+            if (!_tablet_schema->column(index).is_key()) {
+                reader_columns.push_back(index);
+            }
+        }
+    }
+    // Actually only the key columns need to be sorted by id, here we check all
+    // for simplicity.
+    DCHECK(std::is_sorted(reader_columns.begin(), reader_columns.end()));
+
     return Status::OK();
 }
 
@@ -499,8 +516,8 @@ Status OlapChunkSource::_init_olap_reader(RuntimeState* runtime_state) {
 
     RETURN_IF_ERROR(_init_global_dicts(&_params));
     RETURN_IF_ERROR(_init_unused_output_columns(thrift_olap_scan_node.unused_output_column_name));
-    RETURN_IF_ERROR(_init_scanner_columns(scanner_columns));
-    RETURN_IF_ERROR(_init_reader_params(_scan_ctx->key_ranges(), scanner_columns, reader_columns));
+    RETURN_IF_ERROR(_init_reader_params(_scan_ctx->key_ranges()));
+    RETURN_IF_ERROR(_init_scanner_columns(scanner_columns, reader_columns));
 
     // schema is new object, but fields not
     starrocks::Schema child_schema = ChunkHelper::convert_schema(_tablet_schema, reader_columns);
diff --git a/be/src/exec/pipeline/scan/olap_chunk_source.h b/be/src/exec/pipeline/scan/olap_chunk_source.h
@@ -56,9 +56,8 @@ class OlapChunkSource final : public ChunkSource {
     Status _read_chunk(RuntimeState* state, ChunkPtr* chunk) override;
 
     Status _get_tablet(const TInternalScanRange* scan_range);
-    Status _init_reader_params(const std::vector<std::unique_ptr<OlapScanRange>>& key_ranges,
-                               const std::vector<uint32_t>& scanner_columns, std::vector<uint32_t>& reader_columns);
-    Status _init_scanner_columns(std::vector<uint32_t>& scanner_columns);
+    Status _init_reader_params(const std::vector<std::unique_ptr<OlapScanRange>>& key_ranges);
+    Status _init_scanner_columns(std::vector<uint32_t>& scanner_columns, std::vector<uint32_t>& reader_columns);
     Status _init_unused_output_columns(const std::vector<std::string>& unused_output_columns);
     Status _init_olap_reader(RuntimeState* state);
     TCounterMinMaxType::type _get_counter_min_max_type(const std::string& metric_name);
diff --git a/be/src/storage/rowset/segment_iterator.cpp b/be/src/storage/rowset/segment_iterator.cpp
@@ -1837,8 +1837,10 @@ Status SegmentIterator::_build_context(ScanContext* ctx) {
             ctx->_column_iterators.emplace_back(iter);
             ctx->_is_dict_column.emplace_back(false);
             ctx->_dict_decode_schema.append(f);
-            ctx->_subfield_columns.emplace_back(i);
-            ctx->_subfield_iterators.emplace_back(iter);
+            if (output_columns.count(f->id()) != 0) {
+                ctx->_subfield_columns.emplace_back(i);
+                ctx->_subfield_iterators.emplace_back(iter);
+            }
         } else {
             ctx->_read_schema.append(f);
             ctx->_column_iterators.emplace_back(_column_iterators[cid].get());
diff --git a/fe/fe-core/src/main/java/com/starrocks/sql/plan/PlanFragmentBuilder.java b/fe/fe-core/src/main/java/com/starrocks/sql/plan/PlanFragmentBuilder.java
diff --git a/fe/fe-core/src/test/java/com/starrocks/sql/plan/FilterUnusedColumnTest.java b/fe/fe-core/src/test/java/com/starrocks/sql/plan/FilterUnusedColumnTest.java
