branch-4.1: [opt](paimon) add table cache in paimon jni scanner (#66018)

be/src/exec/connector/jni_connector.cpp

@@ -20,6 +20,7 @@
 #include <glog/logging.h>
 
 #include <sstream>
+#include <utility>
 #include <variant>
 
 #include "core/block/block.h"
@@ -166,43 +167,30 @@ Status JniConnector::get_statistics(JNIEnv* env, std::map<std::string, std::stri
 }
 
 Status JniConnector::close() {
-    if (!_closed) {
-        JNIEnv* env = nullptr;
-        RETURN_IF_ERROR(Jni::Env::Get(&env));
-        if (_scanner_opened) {
-            COUNTER_UPDATE(_open_scanner_time, _jni_scanner_open_watcher);
-            COUNTER_UPDATE(_fill_block_time, _fill_block_watcher);
-
-            RETURN_ERROR_IF_EXC(env);
-            int64_t _append = 0;
-            RETURN_IF_ERROR(
-                    _jni_scanner_obj.call_long_method(env, _jni_scanner_get_append_data_time)
-                            .call(&_append));
-
-            COUNTER_UPDATE(_java_append_data_time, _append);
-
-            int64_t _create = 0;
-            RETURN_IF_ERROR(
-                    _jni_scanner_obj
-                            .call_long_method(env, _jni_scanner_get_create_vector_table_time)
-                            .call(&_create));
-
-            COUNTER_UPDATE(_java_create_vector_table_time, _create);
-
-            COUNTER_UPDATE(_java_scan_time, _java_scan_watcher - _append - _create);
+    if (_closed) {
+        return Status::OK();
+    }
+    if (!_scanner_opened) {
+        _closed = true;
+        return Status::OK();
+    }
 
-            _max_time_split_weight_counter->conditional_update(
-                    _jni_scanner_open_watcher + _fill_block_watcher + _java_scan_watcher,
-                    _self_split_weight);
+    JNIEnv* env = nullptr;
+    RETURN_IF_ERROR(Jni::Env::Get(&env));
 
-            // _fill_block may be failed and returned, we should release table in close.
-            // org.apache.doris.common.jni.JniScanner#releaseTable is idempotent
-            RETURN_IF_ERROR(
-                    _jni_scanner_obj.call_void_method(env, _jni_scanner_release_table).call());
-            RETURN_IF_ERROR(_jni_scanner_obj.call_void_method(env, _jni_scanner_close).call());
-        }
+    // _fill_block may fail before releasing the current Java table. JniScanner::releaseTable()
+    // is idempotent, so close always retries it. Java close must still run when that release
+    // fails, otherwise connector resources such as Paimon's static table-cache lease can leak.
+    auto close_status = _jni_scanner_obj.call_void_method(env, _jni_scanner_release_table).call();
+    auto java_close_status = _jni_scanner_obj.call_void_method(env, _jni_scanner_close).call();
+    if (close_status.ok() && !java_close_status.ok()) {
+        close_status = std::move(java_close_status);
     }
-    return Status::OK();
+    if (close_status.ok()) {
+        _scanner_opened = false;
+        _closed = true;
+    }
+    return close_status;
 }
 
 Status JniConnector::_init_jni_scanner(JNIEnv* env, int batch_size) {
@@ -833,6 +821,35 @@ void JniConnector::_collect_profile_before_close() {
             LOG(WARNING) << "failed to get jni env when collect profile: " << st;
             return;
         }
+        COUNTER_UPDATE(_open_scanner_time, _jni_scanner_open_watcher);
+        COUNTER_UPDATE(_fill_block_time, _fill_block_watcher);
+
+        int64_t append_data_time = 0;
+        auto append_time_status =
+                _jni_scanner_obj.call_long_method(env, _jni_scanner_get_append_data_time)
+                        .call(&append_data_time);
+        int64_t create_vector_table_time = 0;
+        auto create_table_time_status =
+                _jni_scanner_obj.call_long_method(env, _jni_scanner_get_create_vector_table_time)
+                        .call(&create_vector_table_time);
+        if (!append_time_status.ok()) {
+            LOG(WARNING) << "failed to collect JNI append-data time before close: "
+                         << append_time_status;
+        }
+        if (!create_table_time_status.ok()) {
+            LOG(WARNING) << "failed to collect JNI vector-table time before close: "
+                         << create_table_time_status;
+        }
+        if (append_time_status.ok() && create_table_time_status.ok()) {
+            COUNTER_UPDATE(_java_append_data_time, append_data_time);
+            COUNTER_UPDATE(_java_create_vector_table_time, create_vector_table_time);
+            COUNTER_UPDATE(_java_scan_time,
+                           _java_scan_watcher - append_data_time - create_vector_table_time);
+            _max_time_split_weight_counter->conditional_update(
+                    _jni_scanner_open_watcher + _fill_block_watcher + _java_scan_watcher,
+                    _self_split_weight);
+        }
+
         // update scanner metrics
         std::map<std::string, std::string> statistics_result;
         st = get_statistics(env, &statistics_result);

be/src/exec/connector/jni_connector.h

@@ -254,6 +254,12 @@ class JniConnector : public ProfileCollector {
      */
     Status close();
 
+#ifdef BE_TEST
+    const std::map<std::string, std::string>& TEST_scanner_params() const {
+        return _scanner_params;
+    }
+#endif
+
     /**
      * Set column name to block index map from FileScanner to avoid repeated map creation.
      */

be/src/format/table/paimon_jni_reader.cpp

@@ -27,6 +27,8 @@
 #include "runtime/exec_env.h"
 #include "runtime/runtime_state.h"
 #include "util/string_util.h"
+#include "util/uid_util.h"
+
 namespace doris {
 class RuntimeProfile;
 class RuntimeState;
@@ -78,6 +80,11 @@ PaimonJniReader::PaimonJniReader(const std::vector<SlotDescriptor*>& file_slot_d
     if (range_params->__isset.serialized_table) {
         params["serialized_table"] = range_params->serialized_table;
     }
+    params["serialized_table_cache_key"] =
+            range_params->__isset.serialized_table_cache_key &&
+                            !range_params->serialized_table_cache_key.empty()
+                    ? range_params->serialized_table_cache_key
+                    : generate_uuid_string();
     if (range.table_format_params.__isset.table_level_row_count) {
         _remaining_table_level_row_count = range.table_format_params.table_level_row_count;
     } else {

be/src/format/table/paimon_jni_reader.h

@@ -18,6 +18,7 @@
 #pragma once
 
 #include <cstddef>
+#include <map>
 #include <string>
 #include <unordered_map>
 #include <unordered_set>
@@ -60,6 +61,12 @@ class PaimonJniReader : public JniReader {
 
     Status init_reader();
 
+#ifdef BE_TEST
+    const std::map<std::string, std::string>& TEST_scanner_params() const {
+        return _jni_connector->TEST_scanner_params();
+    }
+#endif
+
 private:
     int64_t _remaining_table_level_row_count;
 };

be/src/format_v2/jni/paimon_jni_reader.cpp

@@ -22,6 +22,7 @@
 #include "runtime/exec_env.h"
 #include "runtime/runtime_state.h"
 #include "util/string_util.h"
+#include "util/uid_util.h"
 
 namespace doris::format::paimon {
 namespace {
@@ -94,6 +95,12 @@ Status PaimonJniReader::build_scanner_params(std::map<std::string, std::string>*
     (*params)["paimon_split"] = paimon_params.paimon_split;
     (*params)["paimon_predicate"] = *paimon_predicate;
     (*params)["serialized_table"] = _scan_params->serialized_table;
+    // if old Version FE not have set it, generate uuid in BE, so no need to compatible
+    (*params)["serialized_table_cache_key"] =
+            _scan_params->__isset.serialized_table_cache_key &&
+                            !_scan_params->serialized_table_cache_key.empty()
+                    ? _scan_params->serialized_table_cache_key
+                    : generate_uuid_string();
 
     if (_scan_params->__isset.paimon_options && !_scan_params->paimon_options.empty()) {
         for (const auto& kv : _scan_params->paimon_options) {

be/test/format/table/paimon_jni_reader_test.cpp

@@ -0,0 +1,67 @@
+// 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 "format/table/paimon_jni_reader.h"
+
+#include <gtest/gtest.h>
+
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "gen_cpp/PlanNodes_types.h"
+#include "runtime/runtime_state.h"
+
+namespace doris {
+namespace {
+
+TFileRangeDesc make_legacy_paimon_jni_range() {
+    TFileRangeDesc range;
+    TTableFormatFileDesc table_format_params;
+    table_format_params.__set_table_format_type("paimon");
+    TPaimonFileDesc paimon_params;
+    paimon_params.__set_paimon_split("serialized-split");
+    table_format_params.__set_paimon_params(std::move(paimon_params));
+    range.__set_table_format_params(std::move(table_format_params));
+    return range;
+}
+
+TEST(LegacyPaimonJniReaderTest, GeneratesMissingOrEmptySerializedTableCacheKey) {
+    const auto range = make_legacy_paimon_jni_range();
+    TFileScanRangeParams scan_params;
+    scan_params.__set_serialized_table("serialized-table");
+    scan_params.__set_paimon_predicate("serialized-predicate");
+    RuntimeState state {TQueryOptions(), TQueryGlobals()};
+    const std::vector<SlotDescriptor*> file_slot_descs;
+
+    PaimonJniReader missing_key_reader(file_slot_descs, &state, nullptr, range, &scan_params);
+    const auto& missing_params = missing_key_reader.TEST_scanner_params();
+    EXPECT_EQ(missing_params.at("serialized_table"), "serialized-table");
+    const auto& missing_key = missing_params.at("serialized_table_cache_key");
+    EXPECT_FALSE(missing_key.empty());
+
+    scan_params.__set_serialized_table_cache_key("");
+    PaimonJniReader empty_key_reader(file_slot_descs, &state, nullptr, range, &scan_params);
+    const auto& empty_params = empty_key_reader.TEST_scanner_params();
+    EXPECT_EQ(empty_params.at("serialized_table"), "serialized-table");
+    const auto& empty_key = empty_params.at("serialized_table_cache_key");
+    EXPECT_FALSE(empty_key.empty());
+    EXPECT_NE(missing_key, empty_key);
+}
+
+} // namespace
+} // namespace doris

be/test/format_v2/jni/paimon_jni_reader_test.cpp

@@ -84,6 +84,43 @@ TEST(PaimonJniReaderTest, UsesScanLevelPredicateBeforeLegacySplitPredicate) {
     EXPECT_EQ(params["paimon_predicate"], "scan-predicate");
 }
 
+TEST(PaimonJniReaderTest, ForwardsSerializedTableCacheKey) {
+    auto range = make_paimon_jni_range();
+    range.table_format_params.paimon_params.__set_paimon_predicate("serialized-predicate");
+
+    auto scan_params = make_scan_params();
+    scan_params.__set_serialized_table_cache_key("table-cache-key");
+
+    PaimonJniReader reader;
+    ASSERT_TRUE(init_reader(&reader, &scan_params).ok());
+
+    std::map<std::string, std::string> params;
+    ASSERT_TRUE(build_params(&reader, range, &params).ok());
+    EXPECT_EQ(params["serialized_table_cache_key"], "table-cache-key");
+}
+
+TEST(PaimonJniReaderTest, GeneratesMissingOrEmptySerializedTableCacheKey) {
+    auto range = make_paimon_jni_range();
+    range.table_format_params.paimon_params.__set_paimon_predicate("serialized-predicate");
+    auto scan_params = make_scan_params();
+
+    PaimonJniReader reader;
+    ASSERT_TRUE(init_reader(&reader, &scan_params).ok());
+
+    std::map<std::string, std::string> params;
+    ASSERT_TRUE(build_params(&reader, range, &params).ok());
+    EXPECT_EQ(params["serialized_table"], "serialized-table");
+    const std::string missing_key = params["serialized_table_cache_key"];
+    EXPECT_FALSE(missing_key.empty());
+
+    scan_params.__set_serialized_table_cache_key("");
+    ASSERT_TRUE(build_params(&reader, range, &params).ok());
+    EXPECT_EQ(params["serialized_table"], "serialized-table");
+    const std::string empty_key = params["serialized_table_cache_key"];
+    EXPECT_FALSE(empty_key.empty());
+    EXPECT_NE(missing_key, empty_key);
+}
+
 TEST(PaimonJniReaderTest, FallsBackToLegacySplitPredicateWhenScanPredicateIsMissing) {
     auto range = make_paimon_jni_range();
     range.table_format_params.paimon_params.__set_paimon_predicate("legacy-predicate");

be/test/format_v2/table/paimon_reader_test.cpp

@@ -354,6 +354,7 @@ TFileRangeDesc make_legacy_paimon_native_range(TFileFormatType::type physical_fo
 TFileScanRangeParams make_paimon_jni_scan_params() {
     TFileScanRangeParams scan_params;
     scan_params.__set_serialized_table("serialized-paimon-table");
+    scan_params.__set_serialized_table_cache_key("serialized-paimon-table-cache-key");
     scan_params.__set_paimon_predicate("serialized-paimon-predicate");
     return scan_params;
 }
@@ -943,6 +944,7 @@ TEST(PaimonJniReaderTest, BuildScannerParamsKeepsExplicitIOManagerTempDir) {
     EXPECT_EQ(params["paimon.jni.enable_jni_io_manager"], "true");
     EXPECT_EQ(params["paimon.jni.io_manager.tmp_dir"], "/tmp/explicit-paimon-spill");
     EXPECT_EQ(params["paimon.jni.io_manager.impl_class"], "org.example.CustomIOManager");
+    EXPECT_EQ(params["serialized_table_cache_key"], "serialized-paimon-table-cache-key");
 }
 
 TEST(PaimonJniReaderTest, BuildScannerParamsInjectsStorageRootTmpDirForEnabledIOManager) {