This is an automated email from the ASF dual-hosted git repository.
Gabriel39 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 4e74105f43c [improvement](paimon) Support IOManager for JNI reads
(#65332)
4e74105f43c is described below
commit 4e74105f43c5ce9b52606211d966c305c330c7f2
Author: Gabriel <[email protected]>
AuthorDate: Wed Jul 8 10:32:16 2026 +0800
[improvement](paimon) Support IOManager for JNI reads (#65332)
Paimon JNI reads created `TableRead` without `withIOManager`, so Paimon
primary-key merge reads could not spill through Paimon IOManager. This
PR wires Doris catalog properties through FE and BE into the Java
scanner, creates the Paimon IOManager for JNI reads, and closes it on
scanner close or open failure.
The same IOManager option/default temp-dir handling is implemented in
both Paimon JNI scan paths:
- legacy `be/src/format/table/paimon_jni_reader.cpp`
- Format V2 `be/src/format_v2/jni/paimon_jni_reader.cpp`
---
be/src/format/table/paimon_jni_reader.cpp | 28 +++++
be/src/format/table/paimon_jni_reader.h | 2 +
be/src/format_v2/jni/paimon_jni_reader.cpp | 32 +++++
be/src/format_v2/jni/paimon_jni_reader.h | 18 +++
be/test/format_v2/table/paimon_reader_test.cpp | 80 ++++++++++++
.../org/apache/doris/paimon/PaimonJniScanner.java | 138 ++++++++++++++++++++-
.../apache/doris/paimon/PaimonJniScannerTest.java | 112 ++++++++++++++++-
.../datasource/paimon/source/PaimonScanNode.java | 25 +++-
.../paimon/source/PaimonScanNodeTest.java | 28 +++++
9 files changed, 458 insertions(+), 5 deletions(-)
diff --git a/be/src/format/table/paimon_jni_reader.cpp
b/be/src/format/table/paimon_jni_reader.cpp
index c2d43406504..3c734c377d9 100644
--- a/be/src/format/table/paimon_jni_reader.cpp
+++ b/be/src/format/table/paimon_jni_reader.cpp
@@ -18,11 +18,15 @@
#include "paimon_jni_reader.h"
#include <map>
+#include <string_view>
+#include <vector>
#include "core/types.h"
#include "format/jni/jni_data_bridge.h"
#include "runtime/descriptors.h"
+#include "runtime/exec_env.h"
#include "runtime/runtime_state.h"
+#include "util/string_util.h"
namespace doris {
class RuntimeProfile;
@@ -32,8 +36,14 @@ class Block;
namespace doris {
+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,
@@ -74,6 +84,24 @@ PaimonJniReader::PaimonJniReader(const
std::vector<SlotDescriptor*>& file_slot_d
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, ":");
+ }
if (range_params->__isset.properties &&
!range_params->properties.empty()) {
for (const auto& kv : range_params->properties) {
params[HADOOP_OPTION_PREFIX + kv.first] =
kv.second;
diff --git a/be/src/format/table/paimon_jni_reader.h
b/be/src/format/table/paimon_jni_reader.h
index 06b2cdf50e8..532c36ad727 100644
--- a/be/src/format/table/paimon_jni_reader.h
+++ b/be/src/format/table/paimon_jni_reader.h
@@ -47,6 +47,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);
diff --git a/be/src/format_v2/jni/paimon_jni_reader.cpp
b/be/src/format_v2/jni/paimon_jni_reader.cpp
index c68cc7b952a..3902093b7c8 100644
--- a/be/src/format_v2/jni/paimon_jni_reader.cpp
+++ b/be/src/format_v2/jni/paimon_jni_reader.cpp
@@ -19,11 +19,18 @@
#include <string_view>
+#include "runtime/exec_env.h"
+#include "runtime/runtime_state.h"
+#include "util/string_util.h"
+
namespace doris::format::paimon {
namespace {
constexpr std::string_view PAIMON_OPTION_PREFIX = "paimon.";
constexpr std::string_view HADOOP_OPTION_PREFIX = "hadoop.";
+constexpr std::string_view DORIS_ENABLE_JNI_IO_MANAGER =
"doris.enable_jni_io_manager";
+constexpr std::string_view DORIS_JNI_IO_MANAGER_TMP_DIR =
"doris.jni_io_manager.tmp_dir";
+constexpr std::string_view PAIMON_JNI_SCANNER_IO_TMP_DIR =
"paimon_jni_scanner_io_tmp";
} // namespace
@@ -79,6 +86,20 @@ Status
PaimonJniReader::build_scanner_params(std::map<std::string, std::string>*
(*params)[std::string(PAIMON_OPTION_PREFIX) + kv.first] =
kv.second;
}
}
+ const std::string enable_io_manager_key =
+ std::string(PAIMON_OPTION_PREFIX) +
std::string(DORIS_ENABLE_JNI_IO_MANAGER);
+ const std::string io_manager_tmp_dir_key =
+ std::string(PAIMON_OPTION_PREFIX) +
std::string(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()) {
+ DORIS_CHECK(_runtime_state != nullptr);
+ // Keep Format V2 consistent with the legacy JNI path. Paimon creates
and
+ // removes its own paimon-* child directory under these Doris
storage-root scoped
+ // parent directories.
+ (*params)[io_manager_tmp_dir_key] =
+
build_default_io_manager_tmp_dirs(_runtime_state->exec_env()->store_paths());
+ }
if (_scan_params->__isset.properties && !_scan_params->properties.empty())
{
for (const auto& kv : _scan_params->properties) {
(*params)[std::string(HADOOP_OPTION_PREFIX) + kv.first] =
kv.second;
@@ -90,4 +111,15 @@ Status
PaimonJniReader::build_scanner_params(std::map<std::string, std::string>*
return Status::OK();
}
+std::string PaimonJniReader::build_default_io_manager_tmp_dirs(
+ const std::vector<StorePath>& store_paths) {
+ std::vector<std::string> tmp_dirs;
+ tmp_dirs.reserve(store_paths.size());
+ for (const auto& store_path : store_paths) {
+ tmp_dirs.push_back(store_path.path + "/" +
std::string(PAIMON_JNI_SCANNER_IO_TMP_DIR));
+ }
+ DORIS_CHECK(!tmp_dirs.empty());
+ return join(tmp_dirs, ":");
+}
+
} // namespace doris::format::paimon
diff --git a/be/src/format_v2/jni/paimon_jni_reader.h
b/be/src/format_v2/jni/paimon_jni_reader.h
index f789edb0b17..30788081565 100644
--- a/be/src/format_v2/jni/paimon_jni_reader.h
+++ b/be/src/format_v2/jni/paimon_jni_reader.h
@@ -19,11 +19,13 @@
#include <map>
#include <string>
+#include <utility>
#include <vector>
#include "common/status.h"
#include "format_v2/jni/jni_table_reader.h"
#include "gen_cpp/PlanNodes_types.h"
+#include "storage/options.h"
namespace doris::format::paimon {
@@ -31,10 +33,26 @@ class PaimonJniReader final : public format::JniTableReader
{
public:
~PaimonJniReader() override = default;
+#ifdef BE_TEST
+ void TEST_set_scan_params(TFileScanRangeParams* params) { _scan_params =
params; }
+ void TEST_set_runtime_state(RuntimeState* state) { _runtime_state = state;
}
+ void TEST_set_current_range(TFileRangeDesc range) { _current_range =
std::move(range); }
+ Status TEST_build_scanner_params(std::map<std::string, std::string>*
params) const {
+ return build_scanner_params(params);
+ }
+ static std::string TEST_build_default_io_manager_tmp_dirs(
+ const std::vector<StorePath>& store_paths) {
+ return build_default_io_manager_tmp_dirs(store_paths);
+ }
+#endif
+
protected:
std::string connector_class() const override;
Status validate_scan_range(const TFileRangeDesc& range) const override;
Status build_scanner_params(std::map<std::string, std::string>* params)
const override;
+
+private:
+ static std::string build_default_io_manager_tmp_dirs(const
std::vector<StorePath>& store_paths);
};
} // namespace doris::format::paimon
diff --git a/be/test/format_v2/table/paimon_reader_test.cpp
b/be/test/format_v2/table/paimon_reader_test.cpp
index caa4de6ce44..57dec5e4b2d 100644
--- a/be/test/format_v2/table/paimon_reader_test.cpp
+++ b/be/test/format_v2/table/paimon_reader_test.cpp
@@ -47,12 +47,15 @@
#include "format/table/deletion_vector_reader.h"
#include "format/table/paimon_reader.h"
#include "format_v2/column_data.h"
+#include "format_v2/jni/paimon_jni_reader.h"
#include "gen_cpp/ExternalTableSchema_types.h"
#include "gen_cpp/PlanNodes_types.h"
#include "io/io_common.h"
#include "roaring/roaring.hh"
+#include "runtime/exec_env.h"
#include "runtime/runtime_profile.h"
#include "runtime/runtime_state.h"
+#include "storage/options.h"
namespace doris::format {
namespace {
@@ -306,6 +309,39 @@ TFileRangeDesc
make_paimon_range_without_reader_type(TFileFormatType::type forma
return range;
}
+TFileScanRangeParams make_paimon_jni_scan_params() {
+ TFileScanRangeParams scan_params;
+ scan_params.__set_serialized_table("serialized-paimon-table");
+ scan_params.__set_paimon_predicate("serialized-paimon-predicate");
+ return scan_params;
+}
+
+std::map<std::string, std::string> build_paimon_jni_scanner_params(
+ TFileScanRangeParams* scan_params, RuntimeState* state) {
+ paimon::PaimonJniReader reader;
+ reader.TEST_set_scan_params(scan_params);
+ reader.TEST_set_runtime_state(state);
+ reader.TEST_set_current_range(make_paimon_jni_range());
+ std::map<std::string, std::string> params;
+ EXPECT_TRUE(reader.TEST_build_scanner_params(¶ms).ok());
+ return params;
+}
+
+class ScopedExecEnvStorePaths {
+public:
+ explicit ScopedExecEnvStorePaths(std::vector<StorePath> store_paths) {
+ _current =
&const_cast<std::vector<StorePath>&>(ExecEnv::GetInstance()->store_paths());
+ _previous = *_current;
+ *_current = std::move(store_paths);
+ }
+
+ ~ScopedExecEnvStorePaths() { *_current = std::move(_previous); }
+
+private:
+ std::vector<StorePath>* _current = nullptr;
+ std::vector<StorePath> _previous;
+};
+
// Scenario: PaimonReader shares Hudi's history-schema annotation path. A
split whose schema id
// resolves to a historical schema should use field-id mapping and annotate
array/map children so
// TableColumnMapper can match evolved physical Parquet columns by id instead
of by the old names.
@@ -629,5 +665,49 @@ TEST(PaimonHybridReaderTest,
DispatchesNativeThenJniSplitToMatchingReader) {
ASSERT_TRUE(reader.close().ok());
}
+TEST(PaimonJniReaderTest, BuildScannerParamsKeepsExplicitIOManagerTempDir) {
+ auto scan_params = make_paimon_jni_scan_params();
+ scan_params.__set_paimon_options({
+ {"doris.enable_jni_io_manager", "true"},
+ {"doris.jni_io_manager.tmp_dir", "/tmp/explicit-paimon-spill"},
+ {"doris.jni_io_manager.impl_class", "org.example.CustomIOManager"},
+ });
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ state.set_exec_env(ExecEnv::GetInstance());
+
+ auto params = build_paimon_jni_scanner_params(&scan_params, &state);
+ EXPECT_EQ(params["paimon.doris.enable_jni_io_manager"], "true");
+ EXPECT_EQ(params["paimon.doris.jni_io_manager.tmp_dir"],
"/tmp/explicit-paimon-spill");
+ EXPECT_EQ(params["paimon.doris.jni_io_manager.impl_class"],
"org.example.CustomIOManager");
+}
+
+TEST(PaimonJniReaderTest,
BuildScannerParamsInjectsStorageRootTmpDirForEnabledIOManager) {
+ ScopedExecEnvStorePaths store_paths({
+ StorePath("/data1/doris", -1),
+ StorePath("/data2/doris", -1),
+ });
+ auto scan_params = make_paimon_jni_scan_params();
+ scan_params.__set_paimon_options({
+ {"doris.enable_jni_io_manager", "true"},
+ });
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ state.set_exec_env(ExecEnv::GetInstance());
+
+ auto params = build_paimon_jni_scanner_params(&scan_params, &state);
+ EXPECT_EQ(params["paimon.doris.enable_jni_io_manager"], "true");
+ EXPECT_EQ(params["paimon.doris.jni_io_manager.tmp_dir"],
+ "/data1/doris/paimon_jni_scanner_io_tmp:/data2/doris/"
+ "paimon_jni_scanner_io_tmp");
+}
+
+TEST(PaimonJniReaderTest,
BuildScannerParamsUsesStorageRootTmpDirWhenIOManagerTempDirMissing) {
+ std::vector<StorePath> paths;
+ paths.emplace_back("/data1/doris", -1);
+ paths.emplace_back("/data2/doris", -1);
+
EXPECT_EQ(paimon::PaimonJniReader::TEST_build_default_io_manager_tmp_dirs(paths),
+ "/data1/doris/paimon_jni_scanner_io_tmp:/data2/doris/"
+ "paimon_jni_scanner_io_tmp");
+}
+
} // namespace
} // namespace doris::format
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 cdb3d1a7ed0..4593439ec4c 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
@@ -26,19 +26,27 @@ import
org.apache.doris.common.security.authentication.PreExecutionAuthenticator
import com.google.common.base.Preconditions;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.disk.IOManager;
+import org.apache.paimon.disk.IOManagerImpl;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.table.Table;
import org.apache.paimon.table.source.ReadBuilder;
import org.apache.paimon.table.source.Split;
+import org.apache.paimon.table.source.TableRead;
import org.apache.paimon.types.DataType;
import org.apache.paimon.types.TimestampType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
+import java.lang.reflect.Constructor;
+import java.lang.reflect.InvocationTargetException;
+import java.nio.file.Files;
+import java.nio.file.Paths;
import java.util.Arrays;
import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.TimeZone;
@@ -47,6 +55,9 @@ import java.util.stream.Collectors;
public class PaimonJniScanner extends JniScanner {
private static final Logger LOG =
LoggerFactory.getLogger(PaimonJniScanner.class);
private static final String HADOOP_OPTION_PREFIX = "hadoop.";
+ static final String ENABLE_JNI_IO_MANAGER =
"paimon.doris.enable_jni_io_manager";
+ static final String JNI_IO_MANAGER_TMP_DIR =
"paimon.doris.jni_io_manager.tmp_dir";
+ static final String JNI_IO_MANAGER_IMPL_CLASS =
"paimon.doris.jni_io_manager.impl_class";
private final Map<String, String> params;
private final Map<String, String> hadoopOptionParams;
@@ -54,6 +65,8 @@ public class PaimonJniScanner extends JniScanner {
private final String paimonPredicate;
private Table table;
private RecordReader<InternalRow> reader;
+ private IOManager ioManager;
+ private String ioManagerTempDirs;
private final PaimonColumnValue columnValue = new PaimonColumnValue();
private List<String> paimonAllFieldNames;
private List<DataType> paimonDataTypeList;
@@ -105,6 +118,11 @@ public class PaimonJniScanner extends JniScanner {
resetDatetimeV2Precision();
} catch (Throwable e) {
+ try {
+ close();
+ } catch (IOException closeException) {
+ e.addSuppressed(closeException);
+ }
LOG.warn("Failed to open paimon_scanner: " + e.getMessage(), e);
throw new RuntimeException(e);
}
@@ -122,11 +140,96 @@ public class PaimonJniScanner extends JniScanner {
int[] projected = getProjected();
readBuilder.withProjection(projected);
readBuilder.withFilter(getPredicates());
- reader =
readBuilder.newRead().executeFilter().createReader(getSplit());
+ reader =
newReadWithOptionalIOManager(readBuilder).executeFilter().createReader(getSplit());
paimonDataTypeList =
Arrays.stream(projected).mapToObj(i ->
table.rowType().getTypeAt(i)).collect(Collectors.toList());
}
+ private TableRead newReadWithOptionalIOManager(ReadBuilder readBuilder)
throws IOException {
+ TableRead tableRead = readBuilder.newRead();
+ if (!isIOManagerEnabled(params)) {
+ return tableRead;
+ }
+ ioManagerTempDirs = getIOManagerTempDirs(params);
+ ioManager = createIOManager(ioManagerTempDirs,
getIOManagerImplClass(params));
+ LOG.info("Enable Paimon JNI IOManager with temp dirs: {},
implementation: {}",
+ ioManagerTempDirs, ioManager.getClass().getName());
+ return tableRead.withIOManager(ioManager);
+ }
+
+ static boolean isIOManagerEnabled(Map<String, String> params) {
+ return Boolean.parseBoolean(params.getOrDefault(ENABLE_JNI_IO_MANAGER,
"false"));
+ }
+
+ static String getIOManagerTempDirs(Map<String, String> params) throws
IOException {
+ String tempDirs = params.get(JNI_IO_MANAGER_TMP_DIR);
+ if (tempDirs == null || tempDirs.trim().isEmpty()) {
+ throw new IOException("Paimon JNI IOManager is enabled but " +
JNI_IO_MANAGER_TMP_DIR + " is not set");
+ }
+ return tempDirs.trim();
+ }
+
+ static String getIOManagerImplClass(Map<String, String> params) {
+ String implClass = params.get(JNI_IO_MANAGER_IMPL_CLASS);
+ return implClass == null || implClass.trim().isEmpty() ? null :
implClass.trim();
+ }
+
+ static IOManager createIOManager(String tempDirs) throws IOException {
+ return createIOManager(tempDirs, null);
+ }
+
+ static IOManager createIOManager(String tempDirs, String implClassName)
throws IOException {
+ String[] splitDirs = IOManagerImpl.splitPaths(tempDirs);
+ if (splitDirs.length == 0) {
+ throw new IOException("Paimon JNI IOManager temp dirs are empty");
+ }
+ for (String splitDir : splitDirs) {
+ Files.createDirectories(Paths.get(splitDir));
+ }
+ if (implClassName == null) {
+ return IOManager.create(splitDirs);
+ }
+ return createCustomIOManager(implClassName, splitDirs, tempDirs);
+ }
+
+ private static IOManager createCustomIOManager(String implClassName,
String[] splitDirs, String tempDirs)
+ throws IOException {
+ ClassLoader loader = Thread.currentThread().getContextClassLoader();
+ if (loader == null) {
+ loader = PaimonJniScanner.class.getClassLoader();
+ }
+ try {
+ Class<?> implClass = Class.forName(implClassName, true, loader);
+ if (!IOManager.class.isAssignableFrom(implClass)) {
+ throw new IOException("Paimon JNI IOManager implementation " +
implClassName
+ + " does not implement " + IOManager.class.getName());
+ }
+ return (IOManager) instantiateCustomIOManager(implClass,
splitDirs, tempDirs);
+ } catch (ClassNotFoundException e) {
+ throw new IOException("Failed to find Paimon JNI IOManager
implementation: " + implClassName, e);
+ } catch (ReflectiveOperationException e) {
+ throw new IOException("Failed to create Paimon JNI IOManager
implementation: " + implClassName, e);
+ }
+ }
+
+ private static Object instantiateCustomIOManager(Class<?> implClass,
String[] splitDirs, String tempDirs)
+ throws ReflectiveOperationException {
+ try {
+ Constructor<?> constructor =
implClass.getConstructor(String[].class);
+ return constructor.newInstance((Object) splitDirs);
+ } catch (NoSuchMethodException e) {
+ try {
+ Constructor<?> constructor =
implClass.getConstructor(String.class);
+ return constructor.newInstance(tempDirs);
+ } catch (NoSuchMethodException stringConstructorMissing) {
+ Constructor<?> constructor = implClass.getConstructor();
+ return constructor.newInstance();
+ }
+ } catch (InvocationTargetException e) {
+ throw e;
+ }
+ }
+
private int[] getProjected() {
return Arrays.stream(fields).mapToInt(fieldName -> {
int index = paimonAllFieldNames.indexOf(fieldName);
@@ -169,8 +272,32 @@ public class PaimonJniScanner extends JniScanner {
@Override
public void close() throws IOException {
+ IOException exception = null;
if (reader != null) {
- reader.close();
+ try {
+ reader.close();
+ } catch (IOException e) {
+ exception = e;
+ } finally {
+ reader = null;
+ }
+ }
+ if (ioManager != null) {
+ try {
+ ioManager.close();
+ } catch (Exception e) {
+ LOG.warn("Failed to close Paimon JNI IOManager, temp dirs:
{}", ioManagerTempDirs, e);
+ if (exception == null) {
+ exception = new IOException(e);
+ } else {
+ exception.addSuppressed(e);
+ }
+ } finally {
+ ioManager = null;
+ }
+ }
+ if (exception != null) {
+ throw exception;
}
}
@@ -233,6 +360,13 @@ public class PaimonJniScanner extends JniScanner {
return null;
}
+ @Override
+ public Map<String, String> getStatistics() {
+ Map<String, String> statistics = new HashMap<>();
+ statistics.put("counter:PaimonJniIOManagerEnabled", ioManager != null
? "1" : "0");
+ return statistics;
+ }
+
private void initTable() {
Preconditions.checkState(params.containsKey("serialized_table"));
table = PaimonUtils.deserialize(params.get("serialized_table"));
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 5b6f1600c7c..1355d5f9844 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
@@ -17,20 +17,130 @@
package org.apache.doris.paimon;
+import org.apache.paimon.disk.BufferFileReader;
+import org.apache.paimon.disk.BufferFileWriter;
+import org.apache.paimon.disk.FileIOChannel;
+import org.apache.paimon.disk.IOManager;
+import org.apache.paimon.disk.IOManagerImpl;
+import org.junit.Assert;
+import org.junit.Rule;
import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import java.io.File;
+import java.lang.reflect.Field;
import java.util.HashMap;
import java.util.Map;
public class PaimonJniScannerTest {
+ @Rule
+ public TemporaryFolder temporaryFolder = new TemporaryFolder();
+
@Test
public void testConstructorAcceptsEmptyProjection() {
+ new PaimonJniScanner(128, createBaseParams());
+ }
+
+ @Test
+ public void testIOManagerOptionHelpers() throws Exception {
+ Map<String, String> params = createBaseParams();
+ Assert.assertFalse(PaimonJniScanner.isIOManagerEnabled(params));
+
+ params.put(PaimonJniScanner.ENABLE_JNI_IO_MANAGER, "true");
+ File tempDir = new File(temporaryFolder.getRoot(),
"paimon-io-manager");
+ params.put(PaimonJniScanner.JNI_IO_MANAGER_TMP_DIR,
tempDir.getAbsolutePath());
+
+ Assert.assertTrue(PaimonJniScanner.isIOManagerEnabled(params));
+ Assert.assertEquals(tempDir.getAbsolutePath(),
PaimonJniScanner.getIOManagerTempDirs(params));
+ Assert.assertNull(PaimonJniScanner.getIOManagerImplClass(params));
+ PaimonJniScanner.createIOManager(tempDir.getAbsolutePath()).close();
+ Assert.assertTrue(tempDir.exists());
+ }
+
+ @Test
+ public void testCreateDefaultAndCustomIOManager() throws Exception {
+ File tempDir = new File(temporaryFolder.getRoot(),
"paimon-io-manager-impl");
+ IOManager defaultIOManager =
PaimonJniScanner.createIOManager(tempDir.getAbsolutePath());
+ Assert.assertTrue(defaultIOManager instanceof IOManagerImpl);
+ defaultIOManager.close();
+
+ Map<String, String> params = createBaseParams();
+ params.put(PaimonJniScanner.JNI_IO_MANAGER_IMPL_CLASS,
TestIOManager.class.getName());
+ Assert.assertEquals(TestIOManager.class.getName(),
PaimonJniScanner.getIOManagerImplClass(params));
+ IOManager customIOManager = PaimonJniScanner.createIOManager(
+ tempDir.getAbsolutePath(),
PaimonJniScanner.getIOManagerImplClass(params));
+ Assert.assertTrue(customIOManager instanceof TestIOManager);
+ Assert.assertArrayEquals(new String[] {tempDir.getAbsolutePath()},
customIOManager.tempDirs());
+ }
+
+ @Test
+ public void testCloseCleansIOManagerTempDirectory() throws Exception {
+ File tempDir = temporaryFolder.newFolder("paimon-io-manager-clean");
+ IOManager ioManager =
PaimonJniScanner.createIOManager(tempDir.getAbsolutePath());
+ FileIOChannel.ID channel = ioManager.createChannel();
+ File spillFile = channel.getPathFile();
+ Assert.assertTrue(spillFile.createNewFile());
+ File spillDir = spillFile.getParentFile();
+ Assert.assertTrue(spillDir.exists());
+
+ PaimonJniScanner scanner = new PaimonJniScanner(128,
createBaseParams());
+ Field ioManagerField =
PaimonJniScanner.class.getDeclaredField("ioManager");
+ ioManagerField.setAccessible(true);
+ ioManagerField.set(scanner, ioManager);
+ Assert.assertEquals("1",
scanner.getStatistics().get("counter:PaimonJniIOManagerEnabled"));
+
+ scanner.close();
+ Assert.assertFalse(spillDir.exists());
+ }
+
+ private Map<String, String> createBaseParams() {
Map<String, String> params = new HashMap<>();
params.put("required_fields", "");
params.put("columns_types", "");
params.put("paimon_split", "");
params.put("paimon_predicate", "");
+ return params;
+ }
+
+ public static class TestIOManager implements IOManager {
+ private final String[] tempDirs;
+
+ public TestIOManager(String[] tempDirs) {
+ this.tempDirs = tempDirs;
+ }
+
+ @Override
+ public FileIOChannel.ID createChannel() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public FileIOChannel.ID createChannel(String channelName) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public String[] tempDirs() {
+ return tempDirs;
+ }
+
+ @Override
+ public FileIOChannel.Enumerator createChannelEnumerator() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public BufferFileWriter createBufferFileWriter(FileIOChannel.ID
channel) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public BufferFileReader createBufferFileReader(FileIOChannel.ID
channel) {
+ throw new UnsupportedOperationException();
+ }
- new PaimonJniScanner(128, params);
+ @Override
+ public void close() {
+ }
}
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonScanNode.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonScanNode.java
index 27698c2d1f9..ad142f08427 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonScanNode.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonScanNode.java
@@ -67,6 +67,7 @@ import org.apache.paimon.table.source.TableScan;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
@@ -92,6 +93,14 @@ public class PaimonScanNode extends FileQueryScanNode {
private static final String DORIS_START_TIMESTAMP = "startTimestamp";
private static final String DORIS_END_TIMESTAMP = "endTimestamp";
private static final String DORIS_INCREMENTAL_BETWEEN_SCAN_MODE =
"incrementalBetweenScanMode";
+ private static final String PAIMON_PROPERTY_PREFIX = "paimon.";
+ private static final String DORIS_ENABLE_JNI_IO_MANAGER =
"doris.enable_jni_io_manager";
+ private static final String DORIS_JNI_IO_MANAGER_TMP_DIR =
"doris.jni_io_manager.tmp_dir";
+ private static final String DORIS_JNI_IO_MANAGER_IMPL_CLASS =
"doris.jni_io_manager.impl_class";
+ private static final List<String> BACKEND_PAIMON_OPTIONS = Arrays.asList(
+ DORIS_ENABLE_JNI_IO_MANAGER,
+ DORIS_JNI_IO_MANAGER_TMP_DIR,
+ DORIS_JNI_IO_MANAGER_IMPL_CLASS);
private static final String PAIMON_BINLOG_SYSTEM_TABLE_TYPE = "binlog";
private static final String PAIMON_AUDIT_LOG_SYSTEM_TABLE_TYPE =
"audit_log";
@@ -516,12 +525,24 @@ public class PaimonScanNode extends FileQueryScanNode {
return Collections.emptyMap();
}
PaimonExternalCatalog catalog = (PaimonExternalCatalog)
source.getCatalog();
+ Map<String, String> backendOptions = new HashMap<>();
+ Map<String, String> catalogProperties =
catalog.getCatalogProperty().getProperties();
+ if (catalogProperties == null) {
+ catalogProperties = Collections.emptyMap();
+ }
+ for (String option : BACKEND_PAIMON_OPTIONS) {
+ String catalogProperty = PAIMON_PROPERTY_PREFIX + option;
+ if (catalogProperties.containsKey(catalogProperty)) {
+ backendOptions.put(option,
catalogProperties.get(catalogProperty));
+ }
+ }
if (!(catalog.getCatalogProperty().getMetastoreProperties() instanceof
PaimonJdbcMetaStoreProperties)) {
- return Collections.emptyMap();
+ return backendOptions;
}
PaimonJdbcMetaStoreProperties jdbcMetaStoreProperties =
(PaimonJdbcMetaStoreProperties)
catalog.getCatalogProperty().getMetastoreProperties();
- return jdbcMetaStoreProperties.getBackendPaimonOptions();
+
backendOptions.putAll(jdbcMetaStoreProperties.getBackendPaimonOptions());
+ return backendOptions;
}
@VisibleForTesting
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/source/PaimonScanNodeTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/source/PaimonScanNodeTest.java
index c5ddc82a3f6..d745ee451da 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/source/PaimonScanNodeTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/source/PaimonScanNodeTest.java
@@ -511,6 +511,34 @@ public class PaimonScanNodeTest {
Assert.assertEquals(2, backendOptions.size());
}
+ @Test
+ public void testGetBackendPaimonOptionsForJniIOManager() {
+ Map<String, String> props = new HashMap<>();
+ props.put("paimon.doris.enable_jni_io_manager", "true");
+ props.put("paimon.doris.jni_io_manager.tmp_dir", "/tmp/doris-paimon");
+ props.put("paimon.doris.jni_io_manager.impl_class",
"org.example.CustomIOManager");
+
+ CatalogProperty catalogProperty = Mockito.mock(CatalogProperty.class);
+ Mockito.when(catalogProperty.getProperties()).thenReturn(props);
+
Mockito.when(catalogProperty.getMetastoreProperties()).thenReturn(Mockito.mock(MetastoreProperties.class));
+
+ PaimonExternalCatalog catalog =
Mockito.mock(PaimonExternalCatalog.class);
+ Mockito.when(catalog.getCatalogProperty()).thenReturn(catalogProperty);
+
+ PaimonSource source = Mockito.mock(PaimonSource.class);
+ Mockito.when(source.getCatalog()).thenReturn(catalog);
+
+ PaimonScanNode node = newTestNode(new PlanNodeId(0), new TupleId(0),
sv);
+ node.setSource(source);
+
+ Map<String, String> backendOptions = node.getBackendPaimonOptions();
+ Assert.assertEquals("true",
backendOptions.get("doris.enable_jni_io_manager"));
+ Assert.assertEquals("/tmp/doris-paimon",
backendOptions.get("doris.jni_io_manager.tmp_dir"));
+ Assert.assertEquals("org.example.CustomIOManager",
+ backendOptions.get("doris.jni_io_manager.impl_class"));
+ Assert.assertEquals(3, backendOptions.size());
+ }
+
@Test
public void testApplyBackendPaimonOptionsAtScanNodeLevel() throws
Exception {
PaimonScanNode node = newTestNode(new PlanNodeId(0), new TupleId(0),
sv);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
