[feature](reader) Optimize Complex Type Column Reading with Column Pruning.

Co-authored-by: 924060929 <[email protected]>
Co-authored-by: Jerry Hu <[email protected]>

Author: 3 people

be/src/http/action/debug_point_action.cpp

@@ -35,7 +35,6 @@ void BaseDebugPointAction::handle(HttpRequest* req) {
                 "Disable debug points. please check config::enable_debug_points");
     }
     std::string result = status.to_json();
-    LOG(INFO) << "handle request result:" << result;
     if (status.ok()) {
         HttpChannel::send_reply(req, HttpStatus::OK, result);
     } else {

be/src/olap/iterators.h

@@ -46,7 +46,6 @@ struct IteratorRowRef;
 namespace segment_v2 {
 struct SubstreamIterator;
 }
-
 class StorageReadOptions {
 public:
     struct KeyRange {
@@ -143,6 +142,9 @@ class StorageReadOptions {
     std::map<ColumnId, size_t> vir_cid_to_idx_in_block;
     std::map<size_t, vectorized::DataTypePtr> vir_col_idx_to_type;
 
+    std::map<int32_t, TColumnAccessPaths> all_access_paths;
+    std::map<int32_t, TColumnAccessPaths> predicate_access_paths;
+
     std::shared_ptr<vectorized::ScoreRuntime> score_runtime;
     CollectionStatisticsPtr collection_statistics;
 

be/src/olap/rowset/beta_rowset_reader.cpp

@@ -103,6 +103,10 @@ Status BetaRowsetReader::get_segment_iterators(RowsetReaderContext* read_context
     _read_options.remaining_conjunct_roots = _read_context->remaining_conjunct_roots;
     _read_options.common_expr_ctxs_push_down = _read_context->common_expr_ctxs_push_down;
     _read_options.virtual_column_exprs = _read_context->virtual_column_exprs;
+
+    _read_options.all_access_paths = _read_context->all_access_paths;
+    _read_options.predicate_access_paths = _read_context->predicate_access_paths;
+
     _read_options.ann_topn_runtime = _read_context->ann_topn_runtime;
     _read_options.vir_cid_to_idx_in_block = _read_context->vir_cid_to_idx_in_block;
     _read_options.vir_col_idx_to_type = _read_context->vir_col_idx_to_type;

be/src/olap/rowset/rowset_reader_context.h

@@ -18,6 +18,8 @@
 #ifndef DORIS_BE_SRC_OLAP_ROWSET_ROWSET_READER_CONTEXT_H
 #define DORIS_BE_SRC_OLAP_ROWSET_ROWSET_READER_CONTEXT_H
 
+#include <vector>
+
 #include "io/io_common.h"
 #include "olap/column_predicate.h"
 #include "olap/olap_common.h"
@@ -90,6 +92,9 @@ struct RowsetReaderContext {
     std::map<ColumnId, size_t> vir_cid_to_idx_in_block;
     std::map<size_t, vectorized::DataTypePtr> vir_col_idx_to_type;
 
+    std::map<int32_t, TColumnAccessPaths> all_access_paths;
+    std::map<int32_t, TColumnAccessPaths> predicate_access_paths;
+
     std::shared_ptr<vectorized::ScoreRuntime> score_runtime;
     CollectionStatisticsPtr collection_statistics;
     std::shared_ptr<segment_v2::AnnTopNRuntime> ann_topn_runtime;

be/src/olap/rowset/segment_v2/column_reader.cpp

@@ -17,6 +17,7 @@
 
 #pragma once
 
+#include <gen_cpp/Descriptors_types.h>
 #include <gen_cpp/segment_v2.pb.h>
 #include <sys/types.h>
 
@@ -69,6 +70,8 @@ class FileReader;
 struct Slice;
 struct StringRef;
 
+using TColumnAccessPaths = std::vector<TColumnAccessPath>;
+
 namespace segment_v2 {
 
 class EncodingInfo;
@@ -271,7 +274,6 @@ class ColumnReader : public MetadataAdder<ColumnReader>,
     Status _calculate_row_ranges(const std::vector<uint32_t>& page_indexes, RowRanges* row_ranges,
                                  const ColumnIteratorOptions& iter_opts);
 
-private:
     int64_t _meta_length;
     FieldType _meta_type;
     FieldType _meta_children_column_type;
@@ -366,8 +368,54 @@ class ColumnIterator {
 
     virtual bool is_all_dict_encoding() const { return false; }
 
+    virtual Status set_access_paths(const TColumnAccessPaths& all_access_paths,
+                                    const TColumnAccessPaths& predicate_access_paths) {
+        if (!predicate_access_paths.empty()) {
+            _reading_flag = ReadingFlag::READING_FOR_PREDICATE;
+        }
+        return Status::OK();
+    }
+
+    void set_column_name(const std::string& column_name) { _column_name = column_name; }
+
+    const std::string& column_name() const { return _column_name; }
+
+    // Since there may be multiple paths with conflicts or overlaps,
+    // we need to define several reading flags:
+    //
+    // NORMAL_READING — Default value, indicating that the column should be read.
+    // SKIP_READING — The column should not be read.
+    // NEED_TO_READ — The column must be read.
+    // READING_FOR_PREDICATE — The column is required for predicate evaluation.
+    //
+    // For example, suppose there are two paths:
+    // - Path 1 specifies that column A needs to be read, so it is marked as NEED_TO_READ.
+    // - Path 2 specifies that the column should not be read, but since it is already marked as NEED_TO_READ,
+    //   it should not be changed to SKIP_READING.
+    enum class ReadingFlag : int {
+        NORMAL_READING,
+        SKIP_READING,
+        NEED_TO_READ,
+        READING_FOR_PREDICATE
+    };
+    void set_reading_flag(ReadingFlag flag) {
+        if (static_cast<int>(flag) > static_cast<int>(_reading_flag)) {
+            _reading_flag = flag;
+        }
+    }
+
+    ReadingFlag reading_flag() const { return _reading_flag; }
+
+    virtual void set_need_to_read() { set_reading_flag(ReadingFlag::NEED_TO_READ); }
+
+    virtual void remove_pruned_sub_iterators() {};
+
 protected:
+    Result<TColumnAccessPaths> _get_sub_access_paths(const TColumnAccessPaths& access_paths);
     ColumnIteratorOptions _opts;
+
+    ReadingFlag _reading_flag {ReadingFlag::NORMAL_READING};
+    std::string _column_name;
 };
 
 // This iterator is used to read column data from file
@@ -504,6 +552,13 @@ class MapFileColumnIterator final : public ColumnIterator {
         return _offsets_iterator->get_current_ordinal();
     }
 
+    Status set_access_paths(const TColumnAccessPaths& all_access_paths,
+                            const TColumnAccessPaths& predicate_access_paths) override;
+
+    void set_need_to_read() override;
+
+    void remove_pruned_sub_iterators() override;
+
 private:
     std::shared_ptr<ColumnReader> _map_reader = nullptr;
     ColumnIteratorUPtr _null_iterator;
@@ -533,6 +588,13 @@ class StructFileColumnIterator final : public ColumnIterator {
         return _sub_column_iterators[0]->get_current_ordinal();
     }
 
+    Status set_access_paths(const TColumnAccessPaths& all_access_paths,
+                            const TColumnAccessPaths& predicate_access_paths) override;
+
+    void set_need_to_read() override;
+
+    void remove_pruned_sub_iterators() override;
+
 private:
     std::shared_ptr<ColumnReader> _struct_reader = nullptr;
     ColumnIteratorUPtr _null_iterator;
@@ -561,6 +623,12 @@ class ArrayFileColumnIterator final : public ColumnIterator {
         return _offset_iterator->get_current_ordinal();
     }
 
+    Status set_access_paths(const TColumnAccessPaths& all_access_paths,
+                            const TColumnAccessPaths& predicate_access_paths) override;
+    void set_need_to_read() override;
+
+    void remove_pruned_sub_iterators() override;
+
 private:
     std::shared_ptr<ColumnReader> _array_reader = nullptr;
     std::unique_ptr<OffsetFileColumnIterator> _offset_iterator;

be/src/olap/rowset/segment_v2/column_reader.h

@@ -17,12 +17,14 @@
 
 #include "olap/rowset/segment_v2/segment.h"
 
+#include <gen_cpp/Descriptors_types.h>
 #include <gen_cpp/PlanNodes_types.h>
 #include <gen_cpp/olap_file.pb.h>
 #include <gen_cpp/segment_v2.pb.h>
 
 #include <cstring>
 #include <memory>
+#include <sstream>
 #include <utility>
 
 #include "cloud/config.h"
@@ -765,6 +767,20 @@ Status Segment::new_column_iterator(const TabletColumn& tablet_column,
                                                sparse_column_cache_ptr));
     } else {
         RETURN_IF_ERROR(reader->new_iterator(iter, &tablet_column, opt));
+        if (opt->all_access_paths.contains(unique_id) ||
+            opt->predicate_access_paths.contains(unique_id)) {
+            const auto& all_access_paths = opt->all_access_paths.contains(unique_id)
+                                                   ? opt->all_access_paths.at(unique_id)
+                                                   : TColumnAccessPaths {};
+            const auto& predicate_access_paths = opt->predicate_access_paths.contains(unique_id)
+                                                         ? opt->predicate_access_paths.at(unique_id)
+                                                         : TColumnAccessPaths {};
+
+            // set column name to apply access paths.
+            (*iter)->set_column_name(tablet_column.name());
+            RETURN_IF_ERROR((*iter)->set_access_paths(all_access_paths, predicate_access_paths));
+            (*iter)->remove_pruned_sub_iterators();
+        }
     }
 
     if (config::enable_column_type_check && !tablet_column.has_path_info() &&

be/src/olap/rowset/segment_v2/segment.cpp

@@ -263,6 +263,9 @@ Status TabletReader::_capture_rs_readers(const ReaderParams& read_params) {
     _reader_context.ann_topn_runtime = read_params.ann_topn_runtime;
 
     _reader_context.condition_cache_digest = read_params.condition_cache_digest;
+    _reader_context.all_access_paths = read_params.all_access_paths;
+    _reader_context.predicate_access_paths = read_params.predicate_access_paths;
+
     return Status::OK();
 }
 

be/src/olap/tablet_reader.cpp

@@ -17,6 +17,7 @@
 
 #pragma once
 
+#include <gen_cpp/Descriptors_types.h>
 #include <gen_cpp/PaloInternalService_types.h>
 #include <gen_cpp/PlanNodes_types.h>
 #include <stddef.h>
@@ -146,6 +147,9 @@ class TabletReader {
         // slots that cast may be eliminated in storage layer
         std::map<std::string, vectorized::DataTypePtr> target_cast_type_for_variants;
 
+        std::map<int32_t, TColumnAccessPaths> all_access_paths;
+        std::map<int32_t, TColumnAccessPaths> predicate_access_paths;
+
         std::vector<RowSetSplits> rs_splits;
         // For unique key table with merge-on-write
         DeleteBitmapPtr delete_bitmap = nullptr;

be/src/olap/tablet_reader.h

@@ -1684,6 +1684,14 @@ vectorized::Block TabletSchema::create_block(
                             tablet_columns_need_convert_null->find(cid) !=
                                     tablet_columns_need_convert_null->end());
         auto data_type = vectorized::DataTypeFactory::instance().create_data_type(col, is_nullable);
+        if (col.type() == FieldType::OLAP_FIELD_TYPE_STRUCT ||
+            col.type() == FieldType::OLAP_FIELD_TYPE_MAP ||
+            col.type() == FieldType::OLAP_FIELD_TYPE_ARRAY) {
+            if (_pruned_columns_data_type.contains(col.unique_id())) {
+                data_type = _pruned_columns_data_type.at(col.unique_id());
+            }
+        }
+
         if (_vir_col_idx_to_unique_id.contains(cid)) {
             block.insert({vectorized::ColumnNothing::create(0), data_type, col.name()});
             VLOG_DEBUG << fmt::format(
@@ -1703,7 +1711,13 @@ vectorized::Block TabletSchema::create_block(bool ignore_dropped_col) const {
         if (ignore_dropped_col && is_dropped_column(*col)) {
             continue;
         }
+
         auto data_type = vectorized::DataTypeFactory::instance().create_data_type(*col);
+        if (col->type() == FieldType::OLAP_FIELD_TYPE_STRUCT) {
+            if (_pruned_columns_data_type.contains(col->unique_id())) {
+                data_type = _pruned_columns_data_type.at(col->unique_id());
+            }
+        }
         block.insert({data_type->create_column(), data_type, col->name()});
     }
     return block;
@@ -1714,6 +1728,11 @@ vectorized::Block TabletSchema::create_block_by_cids(const std::vector<uint32_t>
     for (const auto& cid : cids) {
         const auto& col = *_cols[cid];
         auto data_type = vectorized::DataTypeFactory::instance().create_data_type(col);
+        if (col.type() == FieldType::OLAP_FIELD_TYPE_STRUCT) {
+            if (_pruned_columns_data_type.contains(col.unique_id())) {
+                data_type = _pruned_columns_data_type.at(col.unique_id());
+            }
+        }
         block.insert({data_type->create_column(), data_type, col.name()});
     }
     return block;