[Enhancement](paimon)support native read paimon top level schema change table.

be/src/vec/exec/format/table/paimon_reader.cpp

@@ -38,6 +38,68 @@ PaimonReader::PaimonReader(std::unique_ptr<GenericReader> file_format_reader,
             ADD_CHILD_TIMER(_profile, "DeleteFileReadTime", paimon_profile);
 }
 
+Status PaimonReader::gen_file_col_name(
+        const std::vector<std::string>& read_table_col_names,
+        const std::unordered_map<uint64_t, std::string>& table_col_id_table_name_map,
+        const std::unordered_map<std::string, ColumnValueRangeType>*
+                table_col_name_to_value_range) {
+    // It is a bit similar to iceberg. I will consider integrating it when I write hudi schema change later.
+    _table_col_to_file_col.clear();
+    _file_col_to_table_col.clear();
+
+    if (!_params.__isset.paimon_schema_info) [[unlikely]] {
+        return Status::RuntimeError("miss paimon schema info.");
+    }
+
+    if (!_params.paimon_schema_info.contains(_range.table_format_params.paimon_params.schema_id))
+            [[unlikely]] {
+        return Status::InternalError("miss paimon schema info.");
+    }
+
+    const auto& table_id_to_file_name =
+            _params.paimon_schema_info.at(_range.table_format_params.paimon_params.schema_id);
+    for (auto [table_col_id, file_col_name] : table_id_to_file_name) {
+        if (table_col_id_table_name_map.find(table_col_id) == table_col_id_table_name_map.end()) {
+            continue;
+        }
+        auto& table_col_name = table_col_id_table_name_map.at(table_col_id);
+
+        _table_col_to_file_col.emplace(table_col_name, file_col_name);
+        _file_col_to_table_col.emplace(file_col_name, table_col_name);
+        if (table_col_name != file_col_name) {
+            _has_schema_change = true;
+        }
+    }
+
+    _all_required_col_names.clear();
+    _not_in_file_col_names.clear();
+    for (auto name : read_table_col_names) {
+        auto iter = _table_col_to_file_col.find(name);
+        if (iter == _table_col_to_file_col.end()) {
+            auto name_low = to_lower(name);
+            _all_required_col_names.emplace_back(name_low);
+
+            _table_col_to_file_col.emplace(name, name_low);
+            _file_col_to_table_col.emplace(name_low, name);
+            if (name != name_low) {
+                _has_schema_change = true;
+            }
+        } else {
+            _all_required_col_names.emplace_back(iter->second);
+        }
+    }
+
+    for (auto& it : *table_col_name_to_value_range) {
+        auto iter = _table_col_to_file_col.find(it.first);
+        if (iter == _table_col_to_file_col.end()) {
+            _new_colname_to_value_range.emplace(it.first, it.second);
+        } else {
+            _new_colname_to_value_range.emplace(iter->second, it.second);
+        }
+    }
+    return Status::OK();
+}
+
 Status PaimonReader::init_row_filters() {
     const auto& table_desc = _range.table_format_params.paimon_params;
     if (!table_desc.__isset.deletion_file) {
@@ -107,7 +169,30 @@ Status PaimonReader::init_row_filters() {
 }
 
 Status PaimonReader::get_next_block_inner(Block* block, size_t* read_rows, bool* eof) {
-    return _file_format_reader->get_next_block(block, read_rows, eof);
+    if (_has_schema_change) {
+        for (int i = 0; i < block->columns(); i++) {
+            ColumnWithTypeAndName& col = block->get_by_position(i);
+            auto iter = _table_col_to_file_col.find(col.name);
+            if (iter != _table_col_to_file_col.end()) {
+                col.name = iter->second;
+            }
+        }
+        block->initialize_index_by_name();
+    }
+
+    RETURN_IF_ERROR(_file_format_reader->get_next_block(block, read_rows, eof));
+
+    if (_has_schema_change) {
+        for (int i = 0; i < block->columns(); i++) {
+            ColumnWithTypeAndName& col = block->get_by_position(i);
+            auto iter = _file_col_to_table_col.find(col.name);
+            if (iter != _file_col_to_table_col.end()) {
+                col.name = iter->second;
+            }
+        }
+        block->initialize_index_by_name();
+    }
+    return Status::OK();
 }
 #include "common/compile_check_end.h"
 } // namespace doris::vectorized

be/src/vec/exec/format/table/paimon_reader.h

@@ -31,12 +31,19 @@ class PaimonReader : public TableFormatReader {
     PaimonReader(std::unique_ptr<GenericReader> file_format_reader, RuntimeProfile* profile,
                  RuntimeState* state, const TFileScanRangeParams& params,
                  const TFileRangeDesc& range, io::IOContext* io_ctx);
+
     ~PaimonReader() override = default;
 
     Status init_row_filters() final;
 
     Status get_next_block_inner(Block* block, size_t* read_rows, bool* eof) final;
 
+    Status gen_file_col_name(
+            const std::vector<std::string>& read_table_col_names,
+            const std::unordered_map<uint64_t, std::string>& table_col_id_table_name_map,
+            const std::unordered_map<std::string, ColumnValueRangeType>*
+                    table_col_name_to_value_range);
+
 protected:
     struct PaimonProfile {
         RuntimeProfile::Counter* num_delete_rows;
@@ -45,6 +52,16 @@ class PaimonReader : public TableFormatReader {
     std::vector<int64_t> _delete_rows;
     PaimonProfile _paimon_profile;
 
+    std::unordered_map<std::string, ColumnValueRangeType> _new_colname_to_value_range;
+
+    std::unordered_map<std::string, std::string> _file_col_to_table_col;
+    std::unordered_map<std::string, std::string> _table_col_to_file_col;
+
+    std::vector<std::string> _all_required_col_names;
+    std::vector<std::string> _not_in_file_col_names;
+
+    bool _has_schema_change = false;
+
     virtual void set_delete_rows() = 0;
 };
 
@@ -61,6 +78,24 @@ class PaimonOrcReader final : public PaimonReader {
         (reinterpret_cast<OrcReader*>(_file_format_reader.get()))
                 ->set_position_delete_rowids(&_delete_rows);
     }
+
+    Status init_reader(
+            const std::vector<std::string>& read_table_col_names,
+            const std::unordered_map<uint64_t, std::string>& table_col_id_table_name_map,
+            std::unordered_map<std::string, ColumnValueRangeType>* table_col_name_to_value_range,
+            const VExprContextSPtrs& conjuncts, const TupleDescriptor* tuple_descriptor,
+            const RowDescriptor* row_descriptor,
+            const VExprContextSPtrs* not_single_slot_filter_conjuncts,
+            const std::unordered_map<int, VExprContextSPtrs>* slot_id_to_filter_conjuncts) {
+        RETURN_IF_ERROR(gen_file_col_name(read_table_col_names, table_col_id_table_name_map,
+                                          table_col_name_to_value_range));
+        auto* orc_reader = static_cast<OrcReader*>(_file_format_reader.get());
+        orc_reader->set_table_col_to_file_col(_table_col_to_file_col);
+        return orc_reader->init_reader(&_all_required_col_names, &_new_colname_to_value_range,
+                                       conjuncts, false, tuple_descriptor, row_descriptor,
+                                       not_single_slot_filter_conjuncts,
+                                       slot_id_to_filter_conjuncts);
+    }
 };
 
 class PaimonParquetReader final : public PaimonReader {
@@ -76,6 +111,26 @@ class PaimonParquetReader final : public PaimonReader {
         (reinterpret_cast<ParquetReader*>(_file_format_reader.get()))
                 ->set_delete_rows(&_delete_rows);
     }
+
+    Status init_reader(
+            const std::vector<std::string>& read_table_col_names,
+            const std::unordered_map<uint64_t, std::string>& table_col_id_table_name_map,
+            std::unordered_map<std::string, ColumnValueRangeType>* table_col_name_to_value_range,
+            const VExprContextSPtrs& conjuncts, const TupleDescriptor* tuple_descriptor,
+            const RowDescriptor* row_descriptor,
+            const std::unordered_map<std::string, int>* colname_to_slot_id,
+            const VExprContextSPtrs* not_single_slot_filter_conjuncts,
+            const std::unordered_map<int, VExprContextSPtrs>* slot_id_to_filter_conjuncts) {
+        RETURN_IF_ERROR(gen_file_col_name(read_table_col_names, table_col_id_table_name_map,
+                                          table_col_name_to_value_range));
+        auto* parquet_reader = static_cast<ParquetReader*>(_file_format_reader.get());
+        parquet_reader->set_table_to_file_col_map(_table_col_to_file_col);
+
+        return parquet_reader->init_reader(
+                _all_required_col_names, _not_in_file_col_names, &_new_colname_to_value_range,
+                conjuncts, tuple_descriptor, row_descriptor, colname_to_slot_id,
+                not_single_slot_filter_conjuncts, slot_id_to_filter_conjuncts);
+    }
 };
 #include "common/compile_check_end.h"
 } // namespace doris::vectorized

be/src/vec/exec/scan/file_scanner.cpp

@@ -997,15 +997,14 @@ Status FileScanner::_get_next_reader() {
                 _cur_reader = std::move(iceberg_reader);
             } else if (range.__isset.table_format_params &&
                        range.table_format_params.table_format_type == "paimon") {
-                std::vector<std::string> place_holder;
-                init_status = parquet_reader->init_reader(
-                        _file_col_names, place_holder, _colname_to_value_range,
-                        _push_down_conjuncts, _real_tuple_desc, _default_val_row_desc.get(),
-                        _col_name_to_slot_id, &_not_single_slot_filter_conjuncts,
-                        &_slot_id_to_filter_conjuncts);
                 std::unique_ptr<PaimonParquetReader> paimon_reader =
                         PaimonParquetReader::create_unique(std::move(parquet_reader), _profile,
                                                            _state, *_params, range, _io_ctx.get());
+                init_status = paimon_reader->init_reader(
+                        _file_col_names, _col_id_name_map, _colname_to_value_range,
+                        _push_down_conjuncts, _real_tuple_desc, _default_val_row_desc.get(),
+                        _col_name_to_slot_id, &_not_single_slot_filter_conjuncts,
+                        &_slot_id_to_filter_conjuncts);
                 RETURN_IF_ERROR(paimon_reader->init_row_filters());
                 _cur_reader = std::move(paimon_reader);
             } else {
@@ -1063,12 +1062,13 @@ Status FileScanner::_get_next_reader() {
                 _cur_reader = std::move(iceberg_reader);
             } else if (range.__isset.table_format_params &&
                        range.table_format_params.table_format_type == "paimon") {
-                init_status = orc_reader->init_reader(
-                        &_file_col_names, _colname_to_value_range, _push_down_conjuncts, false,
-                        _real_tuple_desc, _default_val_row_desc.get(),
-                        &_not_single_slot_filter_conjuncts, &_slot_id_to_filter_conjuncts);
                 std::unique_ptr<PaimonOrcReader> paimon_reader = PaimonOrcReader::create_unique(
                         std::move(orc_reader), _profile, _state, *_params, range, _io_ctx.get());
+
+                init_status = paimon_reader->init_reader(
+                        _file_col_names, _col_id_name_map, _colname_to_value_range,
+                        _push_down_conjuncts, _real_tuple_desc, _default_val_row_desc.get(),
+                        &_not_single_slot_filter_conjuncts, &_slot_id_to_filter_conjuncts);
                 RETURN_IF_ERROR(paimon_reader->init_row_filters());
                 _cur_reader = std::move(paimon_reader);
             } else {
@@ -1273,7 +1273,9 @@ Status FileScanner::_init_expr_ctxes() {
         if (slot_info.is_file_slot) {
             _file_slot_descs.emplace_back(it->second);
             _file_col_names.push_back(it->second->col_name());
-            if (it->second->col_unique_id() > 0) {
+            if (it->second->col_unique_id() >= 0) {
+                // Iceberg field unique ID starts from 1, Paimon/Hudi field unique ID starts from 0.
+                // For other data sources, all columns are set to -1.
                 _col_id_name_map.emplace(it->second->col_unique_id(), it->second->col_name());
             }
         } else {

be/test/vec/exec/format/paimon/paimon_schema_change_test.cpp

@@ -0,0 +1,136 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include <gtest/gtest.h>
+
+#include <map>
+#include <string>
+
+#include "io/file_factory.h"
+#include "io/fs/file_reader.h"
+#include "io/io_common.h"
+#include "runtime/runtime_state.h"
+#include "util/runtime_profile.h"
+#include "vec/exec/format/table/paimon_reader.h"
+
+namespace doris::vectorized {
+
+class PaimonMockReader final : public PaimonReader {
+public:
+    PaimonMockReader(std::unique_ptr<GenericReader> file_format_reader, RuntimeProfile* profile,
+                     RuntimeState* state, const TFileScanRangeParams& params,
+                     const TFileRangeDesc& range, io::IOContext* io_ctx)
+            : PaimonReader(std::move(file_format_reader), profile, state, params, range, io_ctx) {};
+    ~PaimonMockReader() final = default;
+
+    void set_delete_rows() final {
+        (reinterpret_cast<OrcReader*>(_file_format_reader.get()))
+                ->set_position_delete_rowids(&_delete_rows);
+    }
+
+    void check() {
+        ASSERT_TRUE(_has_schema_change == true);
+        ASSERT_TRUE(_new_colname_to_value_range.empty());
+        std::unordered_map<std::string, std::string> table_col_to_file_col_ans;
+        table_col_to_file_col_ans["b"] = "map_col";
+        table_col_to_file_col_ans["e"] = "array_col";
+        table_col_to_file_col_ans["d"] = "struct_col";
+        table_col_to_file_col_ans["a"] = "vvv";
+        table_col_to_file_col_ans["c"] = "k";
+        table_col_to_file_col_ans["nonono"] = "nonono";
+        for (auto [table_col, file_col] : table_col_to_file_col_ans) {
+            ASSERT_TRUE(_table_col_to_file_col[table_col] == file_col);
+            ASSERT_TRUE(_file_col_to_table_col[file_col] == table_col);
+        }
+    }
+};
+
+class PaimonReaderTest : public ::testing::Test {
+protected:
+    void SetUp() override {
+        _profile = new RuntimeProfile("test_profile");
+        _state = new RuntimeState(TQueryGlobals());
+        _io_ctx = new io::IOContext();
+        _schema_file_path = "./be/test/exec/test_data/paimon_scanner/schema-0";
+    }
+
+    void TearDown() override {
+        delete _profile;
+        delete _state;
+        delete _io_ctx;
+    }
+
+    RuntimeProfile* _profile;
+    RuntimeState* _state;
+    io::IOContext* _io_ctx;
+    std::string _schema_file_path;
+};
+
+TEST_F(PaimonReaderTest, ReadSchemaFile) {
+    std::map<int64_t, std::string> file_id_to_name;
+    file_id_to_name[0] = "k";
+    file_id_to_name[1] = "vvv";
+    file_id_to_name[2] = "array_col";
+    file_id_to_name[3] = "struct_col";
+    file_id_to_name[6] = "map_col";
+
+    TFileScanRangeParams params;
+    params.file_type = TFileType::FILE_LOCAL;
+    params.properties = {};
+    params.hdfs_params = {};
+    params.__isset.paimon_schema_info = true;
+    params.paimon_schema_info[0] = file_id_to_name;
+    TFileRangeDesc range;
+    range.table_format_params.paimon_params.schema_id = 0;
+
+    PaimonMockReader reader(nullptr, _profile, _state, params, range, _io_ctx);
+
+    //        create table tmp5 (
+    //                k int,
+    //                vVV string,
+    //                array_col array<int>,
+    //                struct_COL struct<a:int,b:string>,
+    //                map_COL map<string,int>
+    //        ) tblproperties (
+    //                'primary-key' = 'k',
+    //                "file.format" = "parquet"
+    //        );
+
+    std::vector<std::string> read_table_col_names;
+    read_table_col_names.emplace_back("a");
+    read_table_col_names.emplace_back("b");
+    read_table_col_names.emplace_back("c");
+    read_table_col_names.emplace_back("d");
+    read_table_col_names.emplace_back("e");
+    read_table_col_names.emplace_back("nonono");
+
+    std::unordered_map<uint64_t, std::string> table_col_id_table_name_map;
+    table_col_id_table_name_map[1] = "a";
+    table_col_id_table_name_map[6] = "b";
+    table_col_id_table_name_map[0] = "c";
+    table_col_id_table_name_map[3] = "d";
+    table_col_id_table_name_map[2] = "e";
+    table_col_id_table_name_map[10] = "nonono";
+
+    std::unordered_map<std::string, ColumnValueRangeType> table_col_name_to_value_range;
+    Status status = reader.gen_file_col_name(read_table_col_names, table_col_id_table_name_map,
+                                             &table_col_name_to_value_range);
+    ASSERT_TRUE(status.ok());
+    reader.check();
+}
+
+} // namespace doris::vectorized