[branch-4.0](cherry-pick) Backport Paimon and Iceberg fixes

be/src/vec/exec/executor/time_sharing/time_sharing_task_executor.cpp

@@ -58,6 +58,24 @@ DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(thread_pool_task_wait_worker_time_ns_total,
                                      MetricUnit::NANOSECONDS);
 DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(thread_pool_task_wait_worker_count_total, MetricUnit::NOUNIT);
 
+namespace {
+
+Result<std::shared_ptr<TimeSharingTaskHandle>> get_time_sharing_task_handle(
+        const std::shared_ptr<TaskHandle>& task_handle, const char* operation) {
+    if (task_handle == nullptr) {
+        return ResultError(Status::InternalError("{} got null task handle", operation));
+    }
+
+    auto handle = std::dynamic_pointer_cast<TimeSharingTaskHandle>(task_handle);
+    if (handle == nullptr) {
+        return ResultError(Status::InternalError("{} got invalid task handle type, task id: {}",
+                                                 operation, task_handle->task_id().to_string()));
+    }
+    return handle;
+}
+
+} // namespace
+
 SplitThreadPoolToken::SplitThreadPoolToken(TimeSharingTaskExecutor* pool,
                                            TimeSharingTaskExecutor::ExecutionMode mode,
                                            std::shared_ptr<SplitQueue> split_queue,
@@ -755,7 +773,7 @@ Status TimeSharingTaskExecutor::add_task(const TaskId& task_id,
 }
 
 Status TimeSharingTaskExecutor::remove_task(std::shared_ptr<TaskHandle> task_handle) {
-    auto handle = std::dynamic_pointer_cast<TimeSharingTaskHandle>(task_handle);
+    auto handle = DORIS_TRY(get_time_sharing_task_handle(task_handle, "remove_task"));
     std::vector<std::shared_ptr<PrioritizedSplitRunner>> splits_to_destroy;
 
     {
@@ -818,7 +836,11 @@ Result<std::vector<SharedListenableFuture<Void>>> TimeSharingTaskExecutor::enque
         }
     }};
     std::vector<SharedListenableFuture<Void>> finished_futures;
-    auto handle = std::dynamic_pointer_cast<TimeSharingTaskHandle>(task_handle);
+    auto handle_result = get_time_sharing_task_handle(task_handle, "enqueue_splits");
+    if (!handle_result.has_value()) {
+        return ResultError(handle_result.error());
+    }
+    auto handle = handle_result.value();
     {
         std::unique_lock<std::mutex> lock(_mutex);
         for (const auto& task_split : splits) {
@@ -851,7 +873,7 @@ Result<std::vector<SharedListenableFuture<Void>>> TimeSharingTaskExecutor::enque
 Status TimeSharingTaskExecutor::re_enqueue_split(std::shared_ptr<TaskHandle> task_handle,
                                                  bool intermediate,
                                                  const std::shared_ptr<SplitRunner>& split) {
-    auto handle = std::dynamic_pointer_cast<TimeSharingTaskHandle>(task_handle);
+    auto handle = DORIS_TRY(get_time_sharing_task_handle(task_handle, "re_enqueue_split"));
     std::shared_ptr<PrioritizedSplitRunner> prioritized_split =
             handle->get_split(split, intermediate);
     prioritized_split->reset_level_priority();

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

@@ -18,10 +18,14 @@
 #include "paimon_jni_reader.h"
 
 #include <map>
+#include <string_view>
+#include <vector>
 
 #include "runtime/descriptors.h"
+#include "runtime/exec_env.h"
 #include "runtime/runtime_state.h"
 #include "runtime/types.h"
+#include "util/string_util.h"
 #include "vec/core/block.h"
 #include "vec/core/types.h"
 namespace doris {
@@ -35,8 +39,14 @@ class Block;
 namespace doris::vectorized {
 #include "common/compile_check_begin.h"
 
+namespace {
+constexpr std::string_view PAIMON_JNI_SCANNER_IO_TMP_DIR = "paimon_jni_scanner_io_tmp";
+} // namespace
+
 const std::string PaimonJniReader::PAIMON_OPTION_PREFIX = "paimon.";
 const std::string PaimonJniReader::HADOOP_OPTION_PREFIX = "hadoop.";
+const std::string PaimonJniReader::DORIS_ENABLE_JNI_IO_MANAGER = "doris.enable_jni_io_manager";
+const std::string PaimonJniReader::DORIS_JNI_IO_MANAGER_TMP_DIR = "doris.jni_io_manager.tmp_dir";
 
 PaimonJniReader::PaimonJniReader(const std::vector<SlotDescriptor*>& file_slot_descs,
                                  RuntimeState* state, RuntimeProfile* profile,
@@ -73,6 +83,20 @@ PaimonJniReader::PaimonJniReader(const std::vector<SlotDescriptor*>& file_slot_d
     for (const auto& kv : range.table_format_params.paimon_params.paimon_options) {
         params[PAIMON_OPTION_PREFIX + kv.first] = kv.second;
     }
+    const std::string enable_io_manager_key = PAIMON_OPTION_PREFIX + DORIS_ENABLE_JNI_IO_MANAGER;
+    const std::string io_manager_tmp_dir_key = PAIMON_OPTION_PREFIX + DORIS_JNI_IO_MANAGER_TMP_DIR;
+    auto enable_io_manager_it = params.find(enable_io_manager_key);
+    if (enable_io_manager_it != params.end() && iequal(enable_io_manager_it->second, "true") &&
+        params.find(io_manager_tmp_dir_key) == params.end()) {
+        std::vector<std::string> tmp_dirs;
+        for (const auto& store_path : state->exec_env()->store_paths()) {
+            tmp_dirs.push_back(store_path.path + "/" + std::string(PAIMON_JNI_SCANNER_IO_TMP_DIR));
+        }
+        DORIS_CHECK(!tmp_dirs.empty());
+        // Paimon's IOManager creates and later removes its own paimon-* child
+        // directory under these Doris storage-root scoped parent directories.
+        params[io_manager_tmp_dir_key] = join(tmp_dirs, ":");
+    }
     // Prefer hadoop conf from scan node level (range_params->properties) over split level
     // to avoid redundant configuration in each split
     if (range_params->__isset.properties && !range_params->properties.empty()) {

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

@@ -50,6 +50,8 @@ class PaimonJniReader : public JniReader {
 public:
     static const std::string PAIMON_OPTION_PREFIX;
     static const std::string HADOOP_OPTION_PREFIX;
+    static const std::string DORIS_ENABLE_JNI_IO_MANAGER;
+    static const std::string DORIS_JNI_IO_MANAGER_TMP_DIR;
     PaimonJniReader(const std::vector<SlotDescriptor*>& file_slot_descs, RuntimeState* state,
                     RuntimeProfile* profile, const TFileRangeDesc& range,
                     const TFileScanRangeParams* range_params);

be/src/vec/exec/jni_connector.cpp

@@ -867,29 +867,59 @@ void JniConnector::_collect_profile_before_close() {
             return;
         }
 
+        const auto update_peak = [](int64_t previous, int64_t current) {
+            return current > previous;
+        };
         for (const auto& metric : statistics_result) {
             std::vector<std::string> type_and_name = split(metric.first, ":");
             if (type_and_name.size() != 2) {
                 LOG(WARNING) << "Name of JNI Scanner metric should be pattern like "
                              << "'metricType:metricName'";
                 continue;
             }
-            long metric_value = std::stol(metric.second);
+            int64_t metric_value = std::stoll(metric.second);
             RuntimeProfile::Counter* scanner_counter;
             if (type_and_name[0] == "timer") {
                 scanner_counter =
                         ADD_CHILD_TIMER(_profile, type_and_name[1], _connector_name.c_str());
+                COUNTER_UPDATE(scanner_counter, metric_value);
             } else if (type_and_name[0] == "counter") {
                 scanner_counter = ADD_CHILD_COUNTER(_profile, type_and_name[1], TUnit::UNIT,
                                                     _connector_name.c_str());
+                COUNTER_UPDATE(scanner_counter, metric_value);
             } else if (type_and_name[0] == "bytes") {
                 scanner_counter = ADD_CHILD_COUNTER(_profile, type_and_name[1], TUnit::BYTES,
                                                     _connector_name.c_str());
+                COUNTER_UPDATE(scanner_counter, metric_value);
+            } else if (type_and_name[0] == "timer_gauge") {
+                scanner_counter =
+                        ADD_CHILD_TIMER(_profile, type_and_name[1], _connector_name.c_str());
+                COUNTER_SET(scanner_counter, metric_value);
+            } else if (type_and_name[0] == "gauge") {
+                scanner_counter = ADD_CHILD_COUNTER(_profile, type_and_name[1], TUnit::UNIT,
+                                                    _connector_name.c_str());
+                COUNTER_SET(scanner_counter, metric_value);
+            } else if (type_and_name[0] == "bytes_gauge") {
+                scanner_counter = ADD_CHILD_COUNTER(_profile, type_and_name[1], TUnit::BYTES,
+                                                    _connector_name.c_str());
+                COUNTER_SET(scanner_counter, metric_value);
+            } else if (type_and_name[0] == "timer_peak") {
+                auto* scanner_peak_counter = _profile->add_conditition_counter(
+                        type_and_name[1], TUnit::TIME_NS, update_peak, _connector_name.c_str());
+                scanner_peak_counter->conditional_update(metric_value, metric_value);
+            } else if (type_and_name[0] == "peak") {
+                auto* scanner_peak_counter = _profile->add_conditition_counter(
+                        type_and_name[1], TUnit::UNIT, update_peak, _connector_name.c_str());
+                scanner_peak_counter->conditional_update(metric_value, metric_value);
+            } else if (type_and_name[0] == "bytes_peak") {
+                auto* scanner_peak_counter = _profile->add_conditition_counter(
+                        type_and_name[1], TUnit::BYTES, update_peak, _connector_name.c_str());
+                scanner_peak_counter->conditional_update(metric_value, metric_value);
             } else {
-                LOG(WARNING) << "Type of JNI Scanner metric should be timer, counter or bytes";
+                LOG(WARNING) << "Type of JNI Scanner metric should be timer, counter, bytes, "
+                             << "timer_gauge, gauge, bytes_gauge, timer_peak, peak or bytes_peak";
                 continue;
             }
-            COUNTER_UPDATE(scanner_counter, metric_value);
         }
     }
 }

be/src/vec/exec/scan/scanner_scheduler.h

@@ -294,13 +294,28 @@ class TaskExecutorSimplifiedScanScheduler final : public ScannerScheduler {
 
     Status submit_scan_task(SimplifiedScanTask scan_task) override {
         if (!_is_stop) {
+            if (scan_task.scanner_context == nullptr) {
+                return Status::InternalError<false>("scanner pool {} got null scanner context.",
+                                                    _sched_name);
+            }
+            if (scan_task.scan_task == nullptr) {
+                return Status::InternalError<false>("scanner pool {} got null scan task.",
+                                                    _sched_name);
+            }
+            auto task_handle = scan_task.scanner_context->task_handle();
+            if (task_handle == nullptr) {
+                return Status::InternalError<false>(
+                        "scanner pool {} got null task handle, scan task first schedule: {}, "
+                        "scanner context: {}",
+                        _sched_name, scan_task.scan_task->is_first_schedule,
+                        scan_task.scanner_context->debug_string());
+            }
             std::shared_ptr<SplitRunner> split_runner;
             if (scan_task.scan_task->is_first_schedule) {
                 split_runner = std::make_shared<ScannerSplitRunner>("scanner_split_runner",
                                                                     scan_task.scan_func);
                 RETURN_IF_ERROR(split_runner->init());
-                auto result = _task_executor->enqueue_splits(
-                        scan_task.scanner_context->task_handle(), false, {split_runner});
+                auto result = _task_executor->enqueue_splits(task_handle, false, {split_runner});
                 if (!result.has_value()) {
                     LOG(WARNING) << "enqueue_splits failed: " << result.error();
                     return result.error();
@@ -311,8 +326,7 @@ class TaskExecutorSimplifiedScanScheduler final : public ScannerScheduler {
                 if (split_runner == nullptr) {
                     return Status::OK();
                 }
-                RETURN_IF_ERROR(_task_executor->re_enqueue_split(
-                        scan_task.scanner_context->task_handle(), false, split_runner));
+                RETURN_IF_ERROR(_task_executor->re_enqueue_split(task_handle, false, split_runner));
             }
             scan_task.scan_task->split_runner = split_runner;
             return Status::OK();

be/test/vec/exec/executor/time_sharing/time_sharing_task_executor_test.cpp

@@ -26,6 +26,7 @@
 #include <future>
 #include <mutex>
 #include <random>
+#include <string>
 #include <thread>
 
 #include "vec/exec/executor/ticker.h"
@@ -290,13 +291,51 @@ class TestingSplitRunner : public SplitRunner {
     ListenableFuture<Void> _completion_future {};
 };
 
+class QueueOnlySplitRunner : public SplitRunner {
+public:
+    Status init() override { return Status::OK(); }
+
+    Result<SharedListenableFuture<Void>> process_for(std::chrono::nanoseconds) override {
+        _started = true;
+        _finished = true;
+        return SharedListenableFuture<Void>::create_ready();
+    }
+
+    void close(const Status& status) override {}
+
+    bool is_finished() override { return _finished.load(); }
+
+    Status finished_status() override { return Status::OK(); }
+
+    std::string get_info() const override { return "queue_only_split"; }
+
+    bool is_started() const { return _started.load(); }
+
+private:
+    std::atomic<bool> _started {false};
+    std::atomic<bool> _finished {false};
+};
+
+class TestingTaskHandle final : public TaskHandle {
+public:
+    explicit TestingTaskHandle(std::string task_id) : _task_id(std::move(task_id)) {}
+
+    Status init() override { return Status::OK(); }
+
+    bool is_closed() const override { return false; }
+
+    TaskId task_id() const override { return _task_id; }
+
+private:
+    TaskId _task_id;
+};
+
 class TimeSharingTaskExecutorTest : public testing::Test {
 protected:
     void SetUp() override {}
 
     void TearDown() override {}
 
-private:
     template <typename Container>
     void assert_split_states(int end_index, const Container& splits) {
         for (int i = 0; i <= end_index; ++i) {
@@ -324,6 +363,48 @@ class TimeSharingTaskExecutorTest : public testing::Test {
     }
 };
 
+TEST_F(TimeSharingTaskExecutorTest, test_invalid_task_handle_returns_error) {
+    auto ticker = std::make_shared<TestingTicker>();
+
+    TimeSharingTaskExecutor::ThreadConfig thread_config;
+    thread_config.thread_name = "invalid_task_handle";
+    thread_config.workload_group = "normal";
+    TimeSharingTaskExecutor executor(thread_config, 0, 1, 1, ticker);
+    ASSERT_TRUE(executor.init().ok());
+
+    auto split = std::make_shared<QueueOnlySplitRunner>();
+
+    auto null_enqueue_result = executor.enqueue_splits(nullptr, false, {split});
+    ASSERT_FALSE(null_enqueue_result.has_value());
+    EXPECT_NE(std::string(null_enqueue_result.error().msg()).find("null task handle"),
+              std::string::npos);
+
+    Status null_re_enqueue_status = executor.re_enqueue_split(nullptr, false, split);
+    ASSERT_FALSE(null_re_enqueue_status.ok());
+    EXPECT_NE(std::string(null_re_enqueue_status.msg()).find("null task handle"),
+              std::string::npos);
+
+    Status null_remove_status = executor.remove_task(nullptr);
+    ASSERT_FALSE(null_remove_status.ok());
+    EXPECT_NE(std::string(null_remove_status.msg()).find("null task handle"), std::string::npos);
+
+    auto invalid_task_handle = std::make_shared<TestingTaskHandle>("invalid_task");
+    auto invalid_enqueue_result = executor.enqueue_splits(invalid_task_handle, false, {split});
+    ASSERT_FALSE(invalid_enqueue_result.has_value());
+    EXPECT_NE(std::string(invalid_enqueue_result.error().msg()).find("invalid task handle type"),
+              std::string::npos);
+
+    Status invalid_re_enqueue_status = executor.re_enqueue_split(invalid_task_handle, false, split);
+    ASSERT_FALSE(invalid_re_enqueue_status.ok());
+    EXPECT_NE(std::string(invalid_re_enqueue_status.msg()).find("invalid task handle type"),
+              std::string::npos);
+
+    Status invalid_remove_status = executor.remove_task(invalid_task_handle);
+    ASSERT_FALSE(invalid_remove_status.ok());
+    EXPECT_NE(std::string(invalid_remove_status.msg()).find("invalid task handle type"),
+              std::string::npos);
+}
+
 TEST_F(TimeSharingTaskExecutorTest, test_tasks_complete) {
     auto ticker = std::make_shared<TestingTicker>();