This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 79cdaefaab1 [opt](paimon) add table cache in paimon jni scanner
(#66018)
79cdaefaab1 is described below
commit 79cdaefaab198a3a1de7875b61aadeb7123e84e2
Author: zhangstar333 <[email protected]>
AuthorDate: Tue Aug 4 16:23:33 2026 +0800
[opt](paimon) add table cache in paimon jni scanner (#66018)
### What problem does this PR solve?
Problem Summary:
in some user case, paimon table have many splits, and if each split
deserialize table, those will cause some memory problem, so add a table
cache in paimon jni scanner
### Release note
None
### Check List (For Author)
- Test <!-- At least one of them must be included. -->
- [ ] Regression test
- [x] Unit Test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason <!-- Add your reason? -->
- Behavior changed:
- [ ] No.
- [ ] Yes. <!-- Explain the behavior change -->
- Does this need documentation?
- [ ] No.
- [ ] Yes. <!-- Add document PR link here. eg:
https://github.com/apache/doris-website/pull/1214 -->
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR
should merge into -->
---
be/src/exec/scan/jdbc_scanner.cpp | 5 +
be/src/exec/scan/jdbc_scanner.h | 1 +
be/src/format/jni/jni_reader.cpp | 87 ++++++++-------
be/src/format/table/paimon_jni_reader.cpp | 6 +
be/src/format_v2/jni/paimon_jni_reader.cpp | 7 ++
be/test/format/table/paimon_jni_reader_test.cpp | 65 +++++++++++
be/test/format_v2/jni/paimon_jni_reader_test.cpp | 37 +++++++
be/test/format_v2/table/paimon_reader_test.cpp | 2 +
.../org/apache/doris/paimon/PaimonJniScanner.java | 54 ++++++++-
.../org/apache/doris/paimon/PaimonTableCache.java | 82 ++++++++++++++
.../apache/doris/paimon/PaimonJniScannerTest.java | 87 ++++++++++++++-
.../apache/doris/paimon/PaimonTableCacheTest.java | 121 +++++++++++++++++++++
.../connector/paimon/PaimonScanPlanProvider.java | 5 +
.../paimon/PaimonScanPlanProviderTest.java | 7 ++
gensrc/thrift/PlanNodes.thrift | 2 +
15 files changed, 519 insertions(+), 49 deletions(-)
diff --git a/be/src/exec/scan/jdbc_scanner.cpp
b/be/src/exec/scan/jdbc_scanner.cpp
index 7c26ed16dcb..0215374220a 100644
--- a/be/src/exec/scan/jdbc_scanner.cpp
+++ b/be/src/exec/scan/jdbc_scanner.cpp
@@ -172,6 +172,11 @@ Status JdbcScanner::_get_block_impl(RuntimeState* state,
Block* block, bool* eof
return Status::OK();
}
+void JdbcScanner::_collect_profile_before_close() {
+ Scanner::_collect_profile_before_close();
+ _jni_reader->collect_profile_before_close();
+}
+
Status JdbcScanner::close(RuntimeState* state) {
if (!_try_close()) {
return Status::OK();
diff --git a/be/src/exec/scan/jdbc_scanner.h b/be/src/exec/scan/jdbc_scanner.h
index 506cc6276f8..856991967be 100644
--- a/be/src/exec/scan/jdbc_scanner.h
+++ b/be/src/exec/scan/jdbc_scanner.h
@@ -71,6 +71,7 @@ public:
protected:
Status _get_block_impl(RuntimeState* state, Block* block, bool* eos)
override;
+ void _collect_profile_before_close() override;
private:
// Build JDBC params from TupleDescriptor for JdbcJniReader
diff --git a/be/src/format/jni/jni_reader.cpp b/be/src/format/jni/jni_reader.cpp
index 546d8446d58..38d511aea9c 100644
--- a/be/src/format/jni/jni_reader.cpp
+++ b/be/src/format/jni/jni_reader.cpp
@@ -214,48 +214,30 @@ Status JniReader::get_table_schema(std::string&
table_schema_str) {
// =========================================================================
Status JniReader::close() {
- if (!_closed) {
+ if (_closed) {
+ return Status::OK();
+ }
+ if (!_scanner_opened) {
_closed = true;
- JNIEnv* env = nullptr;
- RETURN_IF_ERROR(Jni::Env::Get(&env));
- if (_scanner_opened) {
- if (_profile) {
- COUNTER_UPDATE(_open_scanner_time, _jni_scanner_open_watcher);
- COUNTER_UPDATE(_fill_block_time, _fill_block_watcher);
- }
-
- RETURN_ERROR_IF_EXC(env);
- jlong _append = 0;
- RETURN_IF_ERROR(
- _jni_scanner_obj.call_long_method(env,
_jni_scanner_get_append_data_time)
- .call(&_append));
-
- if (_profile) {
- COUNTER_UPDATE(_java_append_data_time, _append);
- }
+ return Status::OK();
+ }
- jlong _create = 0;
- RETURN_IF_ERROR(
- _jni_scanner_obj
- .call_long_method(env,
_jni_scanner_get_create_vector_table_time)
- .call(&_create));
-
- if (_profile) {
- COUNTER_UPDATE(_java_create_vector_table_time, _create);
- COUNTER_UPDATE(_java_scan_time, _java_scan_watcher - _append -
_create);
- _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;
}
// =========================================================================
@@ -416,6 +398,35 @@ void JniReader::_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);
+
+ jlong 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);
+ jlong 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);
diff --git a/be/src/format/table/paimon_jni_reader.cpp
b/be/src/format/table/paimon_jni_reader.cpp
index 15f81fe6fca..9855b808f76 100644
--- a/be/src/format/table/paimon_jni_reader.cpp
+++ b/be/src/format/table/paimon_jni_reader.cpp
@@ -27,6 +27,7 @@
#include "runtime/exec_env.h"
#include "runtime/runtime_state.h"
#include "util/string_util.h"
+#include "util/uid_util.h"
namespace doris {
class RuntimeProfile;
@@ -84,6 +85,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_params->__isset.paimon_options &&
!range_params->paimon_options.empty()) {
for (const auto& kv : range_params->paimon_options) {
diff --git a/be/src/format_v2/jni/paimon_jni_reader.cpp
b/be/src/format_v2/jni/paimon_jni_reader.cpp
index 730d431d4cb..01f33c5cdf0 100644
--- a/be/src/format_v2/jni/paimon_jni_reader.cpp
+++ b/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) {
diff --git a/be/test/format/table/paimon_jni_reader_test.cpp
b/be/test/format/table/paimon_jni_reader_test.cpp
new file mode 100644
index 00000000000..969713429c2
--- /dev/null
+++ b/be/test/format/table/paimon_jni_reader_test.cpp
@@ -0,0 +1,65 @@
+// 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);
+ EXPECT_EQ(missing_key_reader._scanner_params["serialized_table"],
"serialized-table");
+ const auto& missing_key =
missing_key_reader._scanner_params["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);
+ EXPECT_EQ(empty_key_reader._scanner_params["serialized_table"],
"serialized-table");
+ const auto& empty_key =
empty_key_reader._scanner_params["serialized_table_cache_key"];
+ EXPECT_FALSE(empty_key.empty());
+ EXPECT_NE(missing_key, empty_key);
+}
+
+} // namespace
+} // namespace doris
diff --git a/be/test/format_v2/jni/paimon_jni_reader_test.cpp
b/be/test/format_v2/jni/paimon_jni_reader_test.cpp
index 921b6e70ab9..b97e6e6126b 100644
--- a/be/test/format_v2/jni/paimon_jni_reader_test.cpp
+++ b/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, ¶ms).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, ¶ms).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, ¶ms).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");
diff --git a/be/test/format_v2/table/paimon_reader_test.cpp
b/be/test/format_v2/table/paimon_reader_test.cpp
index 32b82ab12ac..8a215961c0e 100644
--- a/be/test/format_v2/table/paimon_reader_test.cpp
+++ b/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) {
diff --git
a/fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonJniScanner.java
b/fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonJniScanner.java
index 01038f243a9..1dbc045d2e4 100644
---
a/fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonJniScanner.java
+++
b/fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonJniScanner.java
@@ -76,11 +76,14 @@ public class PaimonJniScanner extends JniScanner {
private static final String PAIMON_OPTION_PREFIX = "paimon.";
private static final String ASYNC_READER_THREAD_NAME_PREFIX =
"paimon-reader-async-thread";
private static final String FILE_READER_ASYNC_THRESHOLD =
"file-reader-async-threshold";
+ private static final String SERIALIZED_TABLE = "serialized_table";
private static final int MAX_MANIFEST_PARALLELISM = 256;
static final String DORIS_MANIFEST_PARALLELISM_CAP =
"doris.scan.manifest.parallelism-cap";
static final String DORIS_SERIALIZED_SYSTEM_SOURCE =
"doris.serialized-system-source";
static final String DORIS_SYSTEM_TABLE_TYPE = "doris.system-table-type";
+ private static final String SERIALIZED_SYSTEM_SOURCE =
+ PAIMON_OPTION_PREFIX + DORIS_SERIALIZED_SYSTEM_SOURCE;
static final String ENABLE_JNI_IO_MANAGER =
"paimon.jni.enable_jni_io_manager";
static final String JNI_IO_MANAGER_TMP_DIR =
"paimon.jni.io_manager.tmp_dir";
static final String JNI_IO_MANAGER_IMPL_CLASS =
"paimon.jni.io_manager.impl_class";
@@ -95,7 +98,9 @@ public class PaimonJniScanner extends JniScanner {
private final Map<String, String> hadoopOptionParams;
private final String paimonSplit;
private final String paimonPredicate;
+ private final String tableCacheKey;
private Table table;
+ private PaimonTableCache.TableCacheEntry tableCacheEntry;
private RecordReader<InternalRow> reader;
private IOManager ioManager;
private String ioManagerTempDirs;
@@ -134,6 +139,9 @@ public class PaimonJniScanner extends JniScanner {
}
paimonSplit = params.get("paimon_split");
paimonPredicate = params.get("paimon_predicate");
+ tableCacheKey = params.get("serialized_table_cache_key");
+ Preconditions.checkState(tableCacheKey != null &&
!tableCacheKey.isEmpty(),
+ "Missing required Paimon scanner parameter:
serialized_table_cache_key");
String timeZone = params.getOrDefault("time_zone",
TimeZone.getDefault().getID());
columnValue.setTimeZone(timeZone);
initTableInfo(columnTypes, requiredFields, batchSize);
@@ -156,8 +164,7 @@ public class PaimonJniScanner extends JniScanner {
Thread.currentThread().setContextClassLoader(classLoader);
preExecutionAuthenticator.execute(() -> {
PaimonJdbcDriverUtils.registerDriverIfNeeded(params,
classLoader);
- initTable();
- initReader();
+ initTableAndReader();
return null;
});
resetDatetimeV2Precision();
@@ -367,6 +374,7 @@ public class PaimonJniScanner extends JniScanner {
}
}
} finally {
+ releaseCachedTable();
markScannerClosedForMetrics();
}
if (exception != null) {
@@ -638,11 +646,13 @@ public class PaimonJniScanner extends JniScanner {
}
private void initTable() {
- Preconditions.checkState(params.containsKey("serialized_table"));
- table = PaimonUtils.deserialize(params.get("serialized_table"));
- String encodedSystemSource = params.get(PAIMON_OPTION_PREFIX +
DORIS_SERIALIZED_SYSTEM_SOURCE);
+ Preconditions.checkState(params.containsKey(SERIALIZED_TABLE));
+ table = PaimonUtils.deserialize(params.get(SERIALIZED_TABLE));
+ params.remove(SERIALIZED_TABLE);
+ String encodedSystemSource = params.get(SERIALIZED_SYSTEM_SOURCE);
FileStoreTable systemSource = encodedSystemSource == null
? null : PaimonUtils.deserialize(encodedSystemSource);
+ params.remove(SERIALIZED_SYSTEM_SOURCE);
table = applyBackendManifestParallelism(table,
params.get(PAIMON_OPTION_PREFIX +
DORIS_MANIFEST_PARALLELISM_CAP),
Runtime.getRuntime().availableProcessors(), systemSource,
@@ -867,6 +877,40 @@ public class PaimonJniScanner extends JniScanner {
}
}
+ private boolean initTableFromCache() {
+ PaimonTableCache.TableCacheEntry cachedEntry =
PaimonTableCache.acquire(tableCacheKey);
+ if (cachedEntry == null) {
+ return false;
+ }
+ tableCacheEntry = cachedEntry;
+ table = cachedEntry.table();
+ paimonAllFieldNames = cachedEntry.fieldNames();
+ params.remove(SERIALIZED_TABLE);
+ params.remove(SERIALIZED_SYSTEM_SOURCE);
+ return true;
+ }
+
+ private void initTableAndReader() throws IOException {
+ if (initTableFromCache()) {
+ initReader();
+ return;
+ }
+ initTable();
+ initReader();
+ PaimonTableCache.TableCacheEntry candidate =
+ new PaimonTableCache.TableCacheEntry(table,
paimonAllFieldNames);
+ if (PaimonTableCache.publish(tableCacheKey, candidate)) {
+ tableCacheEntry = candidate;
+ }
+ }
+
+ private void releaseCachedTable() {
+ if (tableCacheEntry != null) {
+ PaimonTableCache.release(tableCacheKey, tableCacheEntry);
+ tableCacheEntry = null;
+ }
+ }
+
private static String[] splitParam(String value, String delimiter) {
if (value == null || value.isEmpty()) {
return new String[0];
diff --git
a/fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonTableCache.java
b/fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonTableCache.java
new file mode 100644
index 00000000000..e401f7afe02
--- /dev/null
+++
b/fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonTableCache.java
@@ -0,0 +1,82 @@
+// 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.
+
+package org.apache.doris.paimon;
+
+import com.google.common.base.Preconditions;
+import org.apache.paimon.table.Table;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
+
+final class PaimonTableCache {
+ private static final ConcurrentHashMap<String, TableCacheEntry>
TABLE_CACHE = new ConcurrentHashMap<>();
+
+ private PaimonTableCache() {
+ }
+
+ static TableCacheEntry acquire(String cacheKey) {
+ return TABLE_CACHE.computeIfPresent(cacheKey, (key, entry) -> {
+ entry.users++;
+ return entry;
+ });
+ }
+
+ static boolean publish(String cacheKey, TableCacheEntry entry) {
+ return TABLE_CACHE.putIfAbsent(cacheKey, entry) == null;
+ }
+
+ static void release(String cacheKey, TableCacheEntry expectedEntry) {
+ TABLE_CACHE.compute(cacheKey, (key, currentEntry) -> {
+ Preconditions.checkState(currentEntry == expectedEntry,
+ "Paimon table cache entry changed unexpectedly for key
%s", cacheKey);
+ Preconditions.checkState(currentEntry.users > 0,
+ "Paimon table cache reference count is invalid for key
%s", cacheKey);
+ currentEntry.users--;
+ return currentEntry.users == 0 ? null : currentEntry;
+ });
+ }
+
+ static int size() {
+ return TABLE_CACHE.size();
+ }
+
+ static void clearForTest() {
+ TABLE_CACHE.clear();
+ }
+
+ static final class TableCacheEntry {
+ private final Table table;
+ private final List<String> fieldNames;
+ private int users = 1;
+
+ TableCacheEntry(Table table, List<String> fieldNames) {
+ this.table = table;
+ this.fieldNames = Collections.unmodifiableList(new
ArrayList<>(fieldNames));
+ }
+
+ Table table() {
+ return table;
+ }
+
+ List<String> fieldNames() {
+ return fieldNames;
+ }
+ }
+}
diff --git
a/fe/be-java-extensions/paimon-scanner/src/test/java/org/apache/doris/paimon/PaimonJniScannerTest.java
b/fe/be-java-extensions/paimon-scanner/src/test/java/org/apache/doris/paimon/PaimonJniScannerTest.java
index bd802230e36..8fe6523ed74 100644
---
a/fe/be-java-extensions/paimon-scanner/src/test/java/org/apache/doris/paimon/PaimonJniScannerTest.java
+++
b/fe/be-java-extensions/paimon-scanner/src/test/java/org/apache/doris/paimon/PaimonJniScannerTest.java
@@ -45,6 +45,7 @@ import org.apache.paimon.table.Table;
import org.apache.paimon.table.system.SystemTableLoader;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.InstantiationUtil;
+import org.junit.After;
import org.junit.Assert;
import org.junit.Rule;
import org.junit.Test;
@@ -72,14 +73,63 @@ import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
public class PaimonJniScannerTest {
+ private static final String SERIALIZED_TABLE = "serialized_table";
+ private static final String SERIALIZED_SYSTEM_SOURCE =
+ "paimon.doris.serialized-system-source";
+ private static final String SERIALIZED_TABLE_CACHE_KEY =
"serialized_table_cache_key";
+
@Rule
public TemporaryFolder temporaryFolder = new TemporaryFolder();
+ @After
+ public void clearTableCache() {
+ PaimonTableCache.clearForTest();
+ }
+
@Test
public void testConstructorAcceptsEmptyProjection() {
new PaimonJniScanner(128, createBaseParams());
}
+ @Test
+ public void testConstructorRejectsMissingOrEmptyTableCacheKey() {
+ Map<String, String> params = createBaseParams();
+ params.remove(SERIALIZED_TABLE_CACHE_KEY);
+ assertInvalidTableCacheKey(params);
+
+ params.put(SERIALIZED_TABLE_CACHE_KEY, "");
+ assertInvalidTableCacheKey(params);
+ }
+
+ @Test
+ public void testWarmTableCacheHitReleasesSerializedTablePayloads() throws
Exception {
+ String cacheKey = "warm-table-cache-hit";
+ Map<String, String> params = createBaseParams();
+ params.put(SERIALIZED_TABLE_CACHE_KEY, cacheKey);
+ params.put(SERIALIZED_TABLE, "serialized-table-payload");
+ params.put(SERIALIZED_SYSTEM_SOURCE,
"serialized-system-source-payload");
+ Table cachedTable = newTestTable(Collections.emptyMap());
+ PaimonTableCache.TableCacheEntry cacheEntry =
+ new PaimonTableCache.TableCacheEntry(cachedTable,
Collections.emptyList());
+ Assert.assertTrue(PaimonTableCache.publish(cacheKey, cacheEntry));
+
+ PaimonJniScanner scanner = new PaimonJniScanner(128, params);
+ Method initTableFromCache =
PaimonJniScanner.class.getDeclaredMethod("initTableFromCache");
+ initTableFromCache.setAccessible(true);
+
+ Assert.assertTrue((Boolean) initTableFromCache.invoke(scanner));
+ Assert.assertFalse(params.containsKey(SERIALIZED_TABLE));
+ Assert.assertFalse(params.containsKey(SERIALIZED_SYSTEM_SOURCE));
+ Field tableField = PaimonJniScanner.class.getDeclaredField("table");
+ tableField.setAccessible(true);
+ Assert.assertSame(cachedTable, tableField.get(scanner));
+
+ scanner.close();
+ Assert.assertEquals(1, PaimonTableCache.size());
+ PaimonTableCache.release(cacheKey, cacheEntry);
+ Assert.assertEquals(0, PaimonTableCache.size());
+ }
+
@Test
public void testOldFeSerializedZeroReadBatchIsRejected() throws Exception {
Table configuredTable = (Table)
Proxy.newProxyInstance(Table.class.getClassLoader(),
@@ -638,8 +688,11 @@ public class PaimonJniScannerTest {
}
@Test
- public void testFailedCloseRetainsResourcesForRetry() throws Exception {
- PaimonJniScanner scanner = new PaimonJniScanner(128,
createBaseParams());
+ public void testFailedCloseReleasesCacheAndRetainsResourcesForRetry()
throws Exception {
+ String cacheKey = "retryable-close";
+ Map<String, String> params = createBaseParams();
+ params.put(SERIALIZED_TABLE_CACHE_KEY, cacheKey);
+ PaimonJniScanner scanner = new PaimonJniScanner(128, params);
AtomicInteger iteratorCloseCalls = new AtomicInteger();
RecordReader.RecordIterator<InternalRow> recordIterator =
new RecordReader.RecordIterator<InternalRow>() {
@@ -680,6 +733,13 @@ public class PaimonJniScannerTest {
Field ioManagerField =
PaimonJniScanner.class.getDeclaredField("ioManager");
ioManagerField.setAccessible(true);
ioManagerField.set(scanner, ioManager);
+ PaimonTableCache.TableCacheEntry cacheEntry =
+ new
PaimonTableCache.TableCacheEntry(newTestTable(Collections.emptyMap()),
+ Collections.emptyList());
+ Assert.assertTrue(PaimonTableCache.publish(cacheKey, cacheEntry));
+ Field cacheEntryField =
PaimonJniScanner.class.getDeclaredField("tableCacheEntry");
+ cacheEntryField.setAccessible(true);
+ cacheEntryField.set(scanner, cacheEntry);
try {
scanner.close();
@@ -690,6 +750,7 @@ public class PaimonJniScannerTest {
Assert.assertSame(recordIterator, recordIteratorField.get(scanner));
Assert.assertSame(reader, readerField.get(scanner));
Assert.assertSame(ioManager, ioManagerField.get(scanner));
+ Assert.assertEquals(0, PaimonTableCache.size());
scanner.close();
Assert.assertNull(recordIteratorField.get(scanner));
@@ -698,6 +759,7 @@ public class PaimonJniScannerTest {
Assert.assertEquals(2, iteratorCloseCalls.get());
Assert.assertEquals(2, readerCloseCalls.get());
Assert.assertEquals(2, ioManager.closeCalls.get());
+ Assert.assertEquals(0, PaimonTableCache.size());
}
private Map<String, String> createBaseParams() {
@@ -706,9 +768,19 @@ public class PaimonJniScannerTest {
params.put("columns_types", "");
params.put("paimon_split", "");
params.put("paimon_predicate", "");
+ params.put(SERIALIZED_TABLE_CACHE_KEY, "test-table-cache-key");
return params;
}
+ private void assertInvalidTableCacheKey(Map<String, String> params) {
+ try {
+ new PaimonJniScanner(128, params);
+ Assert.fail("expected constructor to reject an invalid table cache
key");
+ } catch (IllegalStateException e) {
+
Assert.assertTrue(e.getMessage().contains(SERIALIZED_TABLE_CACHE_KEY));
+ }
+ }
+
private String encodeFields(String... fields) {
return Arrays.stream(fields)
.map(field -> "$" +
Base64.getEncoder().encodeToString(field.getBytes(StandardCharsets.UTF_8)))
@@ -716,7 +788,13 @@ public class PaimonJniScannerTest {
}
private void setTableOptions(PaimonJniScanner scanner, Map<String, String>
options) throws Exception {
- Table table = (Table) Proxy.newProxyInstance(
+ Field tableField = PaimonJniScanner.class.getDeclaredField("table");
+ tableField.setAccessible(true);
+ tableField.set(scanner, newTestTable(options));
+ }
+
+ private Table newTestTable(Map<String, String> options) {
+ return (Table) Proxy.newProxyInstance(
Table.class.getClassLoader(), new Class[] {Table.class},
(proxy, method, args) -> {
if ("options".equals(method.getName())) {
return options;
@@ -726,9 +804,6 @@ public class PaimonJniScannerTest {
}
throw new UnsupportedOperationException(method.getName());
});
- Field tableField = PaimonJniScanner.class.getDeclaredField("table");
- tableField.setAccessible(true);
- tableField.set(scanner, table);
}
private static class SerializableTableHandler implements
InvocationHandler, Serializable {
diff --git
a/fe/be-java-extensions/paimon-scanner/src/test/java/org/apache/doris/paimon/PaimonTableCacheTest.java
b/fe/be-java-extensions/paimon-scanner/src/test/java/org/apache/doris/paimon/PaimonTableCacheTest.java
new file mode 100644
index 00000000000..0ec066dc45c
--- /dev/null
+++
b/fe/be-java-extensions/paimon-scanner/src/test/java/org/apache/doris/paimon/PaimonTableCacheTest.java
@@ -0,0 +1,121 @@
+// 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.
+
+package org.apache.doris.paimon;
+
+import org.apache.paimon.table.Table;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.lang.reflect.Proxy;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+public class PaimonTableCacheTest {
+ @After
+ public void tearDown() {
+ PaimonTableCache.clearForTest();
+ }
+
+ @Test
+ public void testAcquireIncrementsAndReleaseRemovesAtZero() {
+ String cacheKey = "reference-count";
+ PaimonTableCache.TableCacheEntry publishedEntry = newEntry();
+ Assert.assertTrue(PaimonTableCache.publish(cacheKey, publishedEntry));
+
+ PaimonTableCache.TableCacheEntry acquiredEntry =
PaimonTableCache.acquire(cacheKey);
+ Assert.assertSame(publishedEntry, acquiredEntry);
+
+ PaimonTableCache.release(cacheKey, acquiredEntry);
+ Assert.assertEquals(1, PaimonTableCache.size());
+
+ PaimonTableCache.release(cacheKey, publishedEntry);
+ Assert.assertEquals(0, PaimonTableCache.size());
+ Assert.assertNull(PaimonTableCache.acquire(cacheKey));
+ }
+
+ @Test
+ public void
testConcurrentAcquireAndReleaseKeepsEntryWhilePublisherUsesIt() throws
Exception {
+ String cacheKey = "concurrent-reference-count";
+ PaimonTableCache.TableCacheEntry publishedEntry = newEntry();
+ Assert.assertTrue(PaimonTableCache.publish(cacheKey, publishedEntry));
+
+ int threadCount = 16;
+ int iterations = 1000;
+ CountDownLatch start = new CountDownLatch(1);
+ ExecutorService executor = Executors.newFixedThreadPool(threadCount);
+ List<Future<?>> futures = IntStream.range(0, threadCount)
+ .mapToObj(ignored -> executor.submit(() -> {
+ start.await();
+ for (int i = 0; i < iterations; i++) {
+ PaimonTableCache.TableCacheEntry entry =
PaimonTableCache.acquire(cacheKey);
+ Assert.assertSame(publishedEntry, entry);
+ PaimonTableCache.release(cacheKey, entry);
+ }
+ return null;
+ }))
+ .collect(Collectors.toList());
+
+ try {
+ start.countDown();
+ for (Future<?> future : futures) {
+ future.get(30, TimeUnit.SECONDS);
+ }
+ } finally {
+ executor.shutdownNow();
+ }
+
+ Assert.assertEquals(1, PaimonTableCache.size());
+ PaimonTableCache.release(cacheKey, publishedEntry);
+ Assert.assertEquals(0, PaimonTableCache.size());
+ }
+
+ @Test
+ public void testOnlyFirstEntryIsPublished() {
+ String cacheKey = "publish-race";
+ PaimonTableCache.TableCacheEntry first = newEntry();
+ PaimonTableCache.TableCacheEntry second = newEntry();
+
+ Assert.assertTrue(PaimonTableCache.publish(cacheKey, first));
+ Assert.assertFalse(PaimonTableCache.publish(cacheKey, second));
+ PaimonTableCache.TableCacheEntry acquiredEntry =
PaimonTableCache.acquire(cacheKey);
+ Assert.assertSame(first, acquiredEntry);
+
+ PaimonTableCache.release(cacheKey, acquiredEntry);
+ PaimonTableCache.release(cacheKey, first);
+ Assert.assertEquals(0, PaimonTableCache.size());
+ }
+
+ private PaimonTableCache.TableCacheEntry newEntry() {
+ Table table = (Table) Proxy.newProxyInstance(
+ Table.class.getClassLoader(), new Class[] {Table.class},
(proxy, method, args) -> {
+ if ("toString".equals(method.getName())) {
+ return "TestPaimonTable";
+ }
+ throw new UnsupportedOperationException(method.getName());
+ });
+ return new PaimonTableCache.TableCacheEntry(table,
Collections.singletonList("field"));
+ }
+}
diff --git
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanPlanProvider.java
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanPlanProvider.java
index 77f5c1a2427..6d2760711fd 100644
---
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanPlanProvider.java
+++
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanPlanProvider.java
@@ -107,6 +107,7 @@ import java.util.Optional;
import java.util.OptionalInt;
import java.util.OptionalLong;
import java.util.Set;
+import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.stream.Collectors;
@@ -206,6 +207,8 @@ public class PaimonScanPlanProvider implements
ConnectorScanPlanProvider {
// Connector-private scan node property key (the engine never reads it):
carries the base64-serialized
// paimon Table from getScanNodeProperties to populateScanLevelParams,
which puts it on the thrift.
private static final String PROP_SERIALIZED_TABLE =
"paimon.serialized_table";
+ private static final String PROP_SERIALIZED_TABLE_CACHE_KEY =
+ "paimon.serialized_table_cache_key";
private static final String DORIS_MANIFEST_PARALLELISM_CAP =
"doris.scan.manifest.parallelism-cap";
private static final String DORIS_SERIALIZED_SYSTEM_SOURCE =
"doris.serialized-system-source";
@@ -905,6 +908,7 @@ public class PaimonScanPlanProvider implements
ConnectorScanPlanProvider {
Table backendTable = tableForBackend(paimonHandle, table);
String serializedTable = encodeObjectToString(backendTable);
props.put(PROP_SERIALIZED_TABLE, serializedTable);
+ props.put(PROP_SERIALIZED_TABLE_CACHE_KEY,
UUID.randomUUID().toString());
OptionalInt backendManifestCap =
backendManifestParallelism(paimonHandle, table);
// Serialized predicates for BE's JNI scanner. ALWAYS emit, even for
the no-filter / empty-predicate
@@ -1790,6 +1794,7 @@ public class PaimonScanPlanProvider implements
ConnectorScanPlanProvider {
String serializedTable = properties.get(PROP_SERIALIZED_TABLE);
if (serializedTable != null) {
params.setSerializedTable(serializedTable);
+
params.setSerializedTableCacheKey(properties.get(PROP_SERIALIZED_TABLE_CACHE_KEY));
}
String predicate = properties.get("paimon.predicate");
diff --git
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonScanPlanProviderTest.java
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonScanPlanProviderTest.java
index 785f82a95a1..31e0aeb73db 100644
---
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonScanPlanProviderTest.java
+++
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonScanPlanProviderTest.java
@@ -2426,6 +2426,13 @@ public class PaimonScanPlanProviderTest {
Assertions.assertTrue(params.isSetSerializedTable(),
"the serialized paimon table must reach the scan-level thrift
params");
Assertions.assertEquals(scanProps.get("paimon.serialized_table"),
params.getSerializedTable());
+ Assertions.assertTrue(params.isSetSerializedTableCacheKey());
+ Assertions.assertFalse(params.getSerializedTableCacheKey().isEmpty());
+
+ TFileScanRangeParams repeatedParams = new TFileScanRangeParams();
+ provider.populateScanLevelParams(repeatedParams, scanProps);
+ Assertions.assertEquals(params.getSerializedTableCacheKey(),
+ repeatedParams.getSerializedTableCacheKey());
}
@Test
diff --git a/gensrc/thrift/PlanNodes.thrift b/gensrc/thrift/PlanNodes.thrift
index bbf2c64b05d..41bf1617585 100644
--- a/gensrc/thrift/PlanNodes.thrift
+++ b/gensrc/thrift/PlanNodes.thrift
@@ -564,6 +564,8 @@ struct TFileScanRangeParams {
// behavior during a BE-first rolling upgrade; version 1 enables file-wide
ID projection and
// logical initial-default materialization.
34: optional i32 iceberg_scan_semantics_version
+ // FE-generated identity for sharing a deserialized table across JNI
scanners in one scan node.
+ 35: optional string serialized_table_cache_key
}
struct TFileRangeDesc {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
