[Feature](catalog) Prune nested columns for hive/iceberg parquet and orc reader.

Author: kaka11chen

.gitmodules

@@ -20,7 +20,7 @@
 [submodule "contrib/apache-orc"]
 	path = contrib/apache-orc
 	url = https://github.com/apache/doris-thirdparty.git
-	branch = orc
+	branch = cq_nested_column_prune_external_table
 [submodule "doris-faiss"]
 	path = contrib/faiss
 	url = https://github.com/apache/doris-thirdparty.git

be/src/runtime/descriptors.cpp

@@ -63,11 +63,11 @@ SlotDescriptor::SlotDescriptor(const TSlotDescriptor& tdesc)
           _is_materialized(tdesc.isMaterialized && tdesc.need_materialize),
           _is_key(tdesc.is_key),
           _column_paths(tdesc.column_paths),
-          _all_access_paths(tdesc.__isset.all_access_paths ? tdesc.all_access_paths
-                                                           : TColumnAccessPaths {}),
-          _predicate_access_paths(tdesc.__isset.predicate_access_paths
-                                          ? tdesc.predicate_access_paths
-                                          : TColumnAccessPaths {}),
+          _all_column_access_paths(tdesc.__isset.all_access_paths ? tdesc.all_access_paths
+                                                                  : TColumnAccessPaths {}),
+          _predicate_column_access_paths(tdesc.__isset.predicate_access_paths
+                                                 ? tdesc.predicate_access_paths
+                                                 : TColumnAccessPaths {}),
           _is_auto_increment(tdesc.__isset.is_auto_increment ? tdesc.is_auto_increment : false),
           _col_default_value(tdesc.__isset.col_default_value ? tdesc.col_default_value : "") {
     if (tdesc.__isset.virtual_column_expr) {

be/src/runtime/descriptors.h

@@ -82,8 +82,10 @@ class SlotDescriptor {
     bool is_key() const { return _is_key; }
     const std::vector<std::string>& column_paths() const { return _column_paths; };
 
-    const TColumnAccessPaths& all_access_paths() const { return _all_access_paths; }
-    const TColumnAccessPaths& predicate_access_paths() const { return _predicate_access_paths; }
+    const TColumnAccessPaths& all_column_access_paths() const { return _all_column_access_paths; }
+    const TColumnAccessPaths& predicate_column_access_paths() const {
+        return _predicate_column_access_paths;
+    }
 
     bool is_auto_increment() const { return _is_auto_increment; }
 
@@ -98,6 +100,10 @@ class SlotDescriptor {
         return virtual_column_expr;
     }
 
+    void set_is_predicate(bool is_predicate) { _is_predicate = is_predicate; }
+
+    bool is_predicate() const { return _is_predicate; }
+
 private:
     friend class DescriptorTbl;
     friend class TupleDescriptor;
@@ -131,14 +137,16 @@ class SlotDescriptor {
     const bool _is_key;
     const std::vector<std::string> _column_paths;
 
-    const TColumnAccessPaths _all_access_paths;
-    const TColumnAccessPaths _predicate_access_paths;
+    const TColumnAccessPaths _all_column_access_paths;
+    const TColumnAccessPaths _predicate_column_access_paths;
 
     const bool _is_auto_increment;
     const std::string _col_default_value;
 
     std::shared_ptr<doris::TExpr> virtual_column_expr = nullptr;
 
+    bool _is_predicate = false;
+
     SlotDescriptor(const TSlotDescriptor& tdesc);
     SlotDescriptor(const PSlotDescriptor& pdesc);
     MOCK_DEFINE(SlotDescriptor();)

be/src/vec/exec/format/orc/vorc_reader.cpp

@@ -112,6 +112,12 @@ struct LazyReadContext {
     std::unordered_map<std::string, VExprContextSPtr> predicate_missing_columns;
     // lazy read missing columns or all missing columns
     std::unordered_map<std::string, VExprContextSPtr> missing_columns;
+
+    std::vector<std::string> partial_predicate_columns;
+
+    // Record the number of rows filled in filter phase for lazy materialization
+    // This is used to check if a column was already processed in filter phase
+    size_t filter_phase_rows = 0;
 };
 
 class OrcReader : public GenericReader {
@@ -158,7 +164,9 @@ class OrcReader : public GenericReader {
             const VExprContextSPtrs* not_single_slot_filter_conjuncts,
             const std::unordered_map<int, VExprContextSPtrs>* slot_id_to_filter_conjuncts,
             std::shared_ptr<TableSchemaChangeHelper::Node> table_info_node_ptr =
-                    TableSchemaChangeHelper::ConstNode::get_instance());
+                    TableSchemaChangeHelper::ConstNode::get_instance(),
+            const std::set<uint64_t>& column_ids = {},
+            const std::set<uint64_t>& filter_column_ids = {});
 
     Status set_fill_columns(
             const std::unordered_map<std::string, std::tuple<std::string, const SlotDescriptor*>>&
@@ -195,7 +203,8 @@ class OrcReader : public GenericReader {
             std::unordered_map<std::string, orc::StringDictionary*>& column_name_to_dict_map,
             bool* is_stripe_filtered);
 
-    static DataTypePtr convert_to_doris_type(const orc::Type* orc_type);
+    DataTypePtr convert_to_doris_type(const orc::Type* orc_type);
+
     static std::string get_field_name_lower_case(const orc::Type* orc_type, int pos);
 
     void set_row_id_column_iterator(
@@ -630,7 +639,8 @@ class OrcReader : public GenericReader {
     size_t _batch_size;
     int64_t _range_start_offset;
     int64_t _range_size;
-    const std::string& _ctz;
+    // const std::string& _ctz;
+    std::string _ctz;
 
     int32_t _offset_days = 0;
     cctz::time_zone _time_zone;
@@ -653,6 +663,12 @@ class OrcReader : public GenericReader {
     // file column name to orc type
     std::unordered_map<std::string, const orc::Type*> _type_map;
 
+    // Optimized type conversion support
+    // column ID to file original type mapping for handling partial column selection
+    std::unordered_map<uint64_t, const orc::Type*> _column_id_to_file_type;
+    // Keep reference to file root type for complete type information
+    const orc::Type* _file_root_type = nullptr;
+
     std::unique_ptr<ORCFileInputStream> _file_input_stream;
     Statistics _statistics;
     OrcProfile _orc_profile;
@@ -673,6 +689,8 @@ class OrcReader : public GenericReader {
     std::vector<DecimalScaleParams> _decimal_scale_params;
     size_t _decimal_scale_params_index;
 
+    std::set<uint64_t> _column_ids;
+    std::set<uint64_t> _filter_column_ids;
     const std::unordered_map<std::string, ColumnValueRangeType>* _colname_to_value_range = nullptr;
     bool _is_acid = false;
     std::unique_ptr<IColumn::Filter> _filter;

be/src/vec/exec/format/orc/vorc_reader.h

@@ -176,6 +176,7 @@ Status FieldDescriptor::parse_node_field(const std::vector<tparquet::SchemaEleme
         parse_physical_field(t_schema, false, child);
 
         node_field->name = t_schema.name;
+        node_field->lower_case_name = to_lower(t_schema.name);
         node_field->data_type = std::make_shared<DataTypeArray>(make_nullable(child->data_type));
         _next_schema_pos = curr_pos + 1;
         node_field->field_id = t_schema.__isset.field_id ? t_schema.field_id : -1;
@@ -193,8 +194,10 @@ Status FieldDescriptor::parse_node_field(const std::vector<tparquet::SchemaEleme
 void FieldDescriptor::parse_physical_field(const tparquet::SchemaElement& physical_schema,
                                            bool is_nullable, FieldSchema* physical_field) {
     physical_field->name = physical_schema.name;
+    physical_field->lower_case_name = to_lower(physical_field->name);
     physical_field->parquet_schema = physical_schema;
     physical_field->physical_type = physical_schema.type;
+    physical_field->column_id = UNASSIGNED_COLUMN_ID; // Initialize column_id
     _physical_fields.push_back(physical_field);
     physical_field->physical_column_index = cast_set<int>(_physical_fields.size() - 1);
     auto type = get_doris_type(physical_schema, is_nullable);
@@ -393,6 +396,8 @@ Status FieldDescriptor::parse_group_field(const std::vector<tparquet::SchemaElem
         RETURN_IF_ERROR(parse_struct_field(t_schemas, curr_pos, struct_field));
 
         group_field->name = group_schema.name;
+        group_field->lower_case_name = to_lower(group_field->name);
+        group_field->column_id = UNASSIGNED_COLUMN_ID; // Initialize column_id
         group_field->data_type =
                 std::make_shared<DataTypeArray>(make_nullable(struct_field->data_type));
         group_field->field_id = group_schema.__isset.field_id ? group_schema.field_id : -1;
@@ -461,6 +466,8 @@ Status FieldDescriptor::parse_list_field(const std::vector<tparquet::SchemaEleme
     }
 
     list_field->name = first_level.name;
+    list_field->lower_case_name = to_lower(first_level.name);
+    list_field->column_id = UNASSIGNED_COLUMN_ID; // Initialize column_id
     list_field->data_type =
             std::make_shared<DataTypeArray>(make_nullable(list_field->children[0].data_type));
     if (is_optional) {
@@ -519,21 +526,24 @@ Status FieldDescriptor::parse_map_field(const std::vector<tparquet::SchemaElemen
     map_field->repetition_level++;
     map_field->definition_level++;
 
-    map_field->children.resize(1);
+    // Directly create key and value children instead of intermediate key_value node
+    map_field->children.resize(2);
     // map is a repeated node, we should set the `repeated_parent_def_level` of its children as `definition_level`
     set_child_node_level(map_field, map_field->definition_level);
-    auto map_kv_field = &map_field->children[0];
-    // produce MAP<STRUCT<KEY, VALUE>>
-    RETURN_IF_ERROR(parse_struct_field(t_schemas, curr_pos + 1, map_kv_field));
+
+    auto key_field = &map_field->children[0];
+    auto value_field = &map_field->children[1];
+
+    // Parse key and value fields directly from the key_value group's children
+    _next_schema_pos = curr_pos + 2; // Skip key_value group, go directly to key
+    RETURN_IF_ERROR(parse_node_field(t_schemas, _next_schema_pos, key_field));
+    RETURN_IF_ERROR(parse_node_field(t_schemas, _next_schema_pos, value_field));
 
     map_field->name = map_schema.name;
-    map_field->data_type = std::make_shared<DataTypeMap>(
-            make_nullable(assert_cast<const DataTypeStruct*>(
-                                  remove_nullable(map_kv_field->data_type).get())
-                                  ->get_element(0)),
-            make_nullable(assert_cast<const DataTypeStruct*>(
-                                  remove_nullable(map_kv_field->data_type).get())
-                                  ->get_element(1)));
+    map_field->lower_case_name = to_lower(map_field->name);
+    map_field->column_id = UNASSIGNED_COLUMN_ID; // Initialize column_id
+    map_field->data_type = std::make_shared<DataTypeMap>(make_nullable(key_field->data_type),
+                                                         make_nullable(value_field->data_type));
     if (is_optional) {
         map_field->data_type = make_nullable(map_field->data_type);
     }
@@ -558,6 +568,8 @@ Status FieldDescriptor::parse_struct_field(const std::vector<tparquet::SchemaEle
         RETURN_IF_ERROR(parse_node_field(t_schemas, _next_schema_pos, &struct_field->children[i]));
     }
     struct_field->name = struct_schema.name;
+    struct_field->lower_case_name = to_lower(struct_field->name);
+    struct_field->column_id = UNASSIGNED_COLUMN_ID; // Initialize column_id
 
     struct_field->field_id = struct_schema.__isset.field_id ? struct_schema.field_id : -1;
     DataTypes res_data_types;
@@ -610,6 +622,59 @@ std::string FieldDescriptor::debug_string() const {
     ss << "]";
     return ss.str();
 }
+
+void FieldDescriptor::assign_ids() {
+    uint64_t next_id = 1;
+    for (auto& field : _fields) {
+        field.assign_ids(next_id);
+    }
+}
+
+const FieldSchema* FieldDescriptor::find_column_by_id(uint64_t column_id) const {
+    for (const auto& field : _fields) {
+        if (auto result = field.find_column_by_id(column_id)) {
+            return result;
+        }
+    }
+    return nullptr;
+}
+
+void FieldSchema::assign_ids(uint64_t& next_id) {
+    column_id = next_id++;
+
+    for (auto& child : children) {
+        child.assign_ids(next_id);
+    }
+
+    max_column_id = next_id - 1;
+}
+
+const FieldSchema* FieldSchema::find_column_by_id(uint64_t target_id) const {
+    if (column_id == target_id) {
+        return this;
+    }
+
+    for (const auto& child : children) {
+        if (auto result = child.find_column_by_id(target_id)) {
+            return result;
+        }
+    }
+
+    return nullptr;
+}
+
+uint64_t FieldSchema::get_column_id() const {
+    return column_id;
+}
+
+void FieldSchema::set_column_id(uint64_t id) {
+    column_id = id;
+}
+
+uint64_t FieldSchema::get_max_column_id() const {
+    return max_column_id;
+}
+
 #include "common/compile_check_end.h"
 
 } // namespace doris::vectorized

be/src/vec/exec/format/parquet/schema_desc.cpp

@@ -36,8 +36,13 @@
 
 namespace doris::vectorized {
 #include "common/compile_check_begin.h"
+
+// Constant for unassigned column IDs
+constexpr uint64_t UNASSIGNED_COLUMN_ID = UINT64_MAX;
+
 struct FieldSchema {
     std::string name;
+    std::string lower_case_name; // for hms column name case insensitive match
     // the referenced parquet schema element
     tparquet::SchemaElement parquet_schema;
 
@@ -56,12 +61,22 @@ struct FieldSchema {
     //For UInt8 -> Int16,UInt16 -> Int32,UInt32 -> Int64,UInt64 -> Int128.
     bool is_type_compatibility = false;
 
-    FieldSchema() : data_type(std::make_shared<DataTypeNothing>()) {}
+    FieldSchema()
+            : data_type(std::make_shared<DataTypeNothing>()), column_id(UNASSIGNED_COLUMN_ID) {}
     ~FieldSchema() = default;
     FieldSchema(const FieldSchema& fieldSchema) = default;
     std::string debug_string() const;
 
     int32_t field_id = -1;
+    uint64_t column_id = UNASSIGNED_COLUMN_ID;
+    uint64_t max_column_id = 0; // Maximum column ID for this field and its children
+
+    // Column ID assignment and lookup methods
+    void assign_ids(uint64_t& next_id);
+    const FieldSchema* find_column_by_id(uint64_t target_id) const;
+    uint64_t get_column_id() const;
+    void set_column_id(uint64_t id);
+    uint64_t get_max_column_id() const;
 };
 
 class FieldDescriptor {
@@ -133,6 +148,19 @@ class FieldDescriptor {
     int32_t size() const { return cast_set<int32_t>(_fields.size()); }
 
     const std::vector<FieldSchema>& get_fields_schema() const { return _fields; }
+
+    /**
+     * Assign stable column IDs to schema fields.
+     *
+     * This uses an ORC-compatible encoding so that the results of
+     * create_column_ids() are consistent across formats. IDs start from 1
+     * and are assigned in a pre-order traversal (parent before children).
+     * After calling this, each FieldSchema will have column_id and
+     * max_column_id populated.
+     */
+    void assign_ids();
+
+    const FieldSchema* find_column_by_id(uint64_t column_id) const;
 };
 #include "common/compile_check_end.h"