[feat](paimon) support paimon-rust reader on BE

be/cmake/thirdparty.cmake

@@ -114,6 +114,11 @@ add_thirdparty(parquet LIB64)
 # so the final linker resolves C math symbols from the system library first.
 add_thirdparty(lance_c LIB64 NOTADD)
 list(APPEND COMMON_THIRDPARTY m lance_c)
+# libpaimon_c.a (built from paimon-rust) also brings in Rust compiler_builtins
+# that would otherwise steal libm symbols. Place libm before it for the same
+# reason as lance_c above.
+add_thirdparty(paimon_c LIB64 NOTADD)
+list(APPEND COMMON_THIRDPARTY m paimon_c)
 add_thirdparty(brpc LIB64)
 add_thirdparty(rocksdb)
 add_thirdparty(cyrus-sasl LIBNAME "lib/libsasl2.a")

be/src/exec/scan/file_scanner.cpp

@@ -78,6 +78,7 @@
 #include "format/table/paimon_jni_reader.h"
 #include "format/table/paimon_predicate_converter.h"
 #include "format/table/paimon_reader.h"
+#include "format/table/paimon_rust_reader.h"
 #include "format/table/partition_column_filler.h"
 #include "format/table/remote_doris_reader.h"
 #include "format/table/transactional_hive_reader.h"
@@ -1097,9 +1098,13 @@ Status FileScanner::_get_next_reader() {
             } else if (range.__isset.table_format_params &&
                        range.table_format_params.table_format_type == "paimon") {
                 const auto& paimon_params = range.table_format_params.paimon_params;
+                bool use_paimon_rust_reader = false;
                 bool use_paimon_cpp_reader = false;
                 if (paimon_params.__isset.reader_type) {
                     switch (paimon_params.reader_type) {
+                    case TPaimonReaderType::PAIMON_RUST:
+                        use_paimon_rust_reader = true;
+                        break;
                     case TPaimonReaderType::PAIMON_CPP:
                         use_paimon_cpp_reader = true;
                         break;
@@ -1116,11 +1121,37 @@ Status FileScanner::_get_next_reader() {
                     }
                 } else {
                     // TODO: Remove this fallback after all FE versions set TPaimonReaderType.
-                    use_paimon_cpp_reader =
-                            _state->query_options().__isset.enable_paimon_cpp_reader &&
-                            _state->query_options().enable_paimon_cpp_reader;
+                    // Rust has higher priority than cpp.
+                    use_paimon_rust_reader =
+                            _state->query_options().__isset.enable_paimon_rust_reader &&
+                            _state->query_options().enable_paimon_rust_reader;
+                    if (!use_paimon_rust_reader) {
+                        use_paimon_cpp_reader =
+                                _state->query_options().__isset.enable_paimon_cpp_reader &&
+                                _state->query_options().enable_paimon_cpp_reader;
+                    }
+                }
+                // Native paimon readers (rust/cpp) push conjuncts down, but they live
+                // under FORMAT_JNI where push_down_predicates is false, so the generic
+                // parquet/orc path never ran _process_late_arrival_conjuncts() and
+                // _push_down_conjuncts is still empty here. Populate it now (same call
+                // parquet/orc make) so the predicate converters see the conjuncts.
+                if (!_is_load && (use_paimon_rust_reader || use_paimon_cpp_reader)) {
+                    RETURN_IF_ERROR(_process_late_arrival_conjuncts());
                 }
-                if (use_paimon_cpp_reader) {
+                if (use_paimon_rust_reader) {
+                    auto rust_reader = PaimonRustReader::create_unique(_file_slot_descs, _state,
+                                                                       _profile, range, _params);
+                    if (!_is_load && !_push_down_conjuncts.empty()) {
+                        // paimon-rust needs the live table handle to build predicates,
+                        // so the converter runs inside init_reader. Just hand over the
+                        // conjuncts here.
+                        rust_reader->set_push_down_conjuncts(_push_down_conjuncts);
+                    }
+                    init_status =
+                            static_cast<GenericReader*>(rust_reader.get())->init_reader(&jni_ctx);
+                    _cur_reader = std::move(rust_reader);
+                } else if (use_paimon_cpp_reader) {
                     auto cpp_reader = PaimonCppReader::create_unique(_file_slot_descs, _state,
                                                                      _profile, range, _params);
                     if (!_is_load && !_push_down_conjuncts.empty()) {