This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.0 by this push:
new 54e9da0b631 [branch-4.0](cherry-pick) Backport Paimon and Iceberg
fixes (#65397)
54e9da0b631 is described below
commit 54e9da0b631ba8e77510d7b1704319d4bc58fa56
Author: Gabriel <[email protected]>
AuthorDate: Tue Jul 14 15:23:26 2026 +0800
[branch-4.0](cherry-pick) Backport Paimon and Iceberg fixes (#65397)
## Proposed changes
Backport the following PRs to `branch-4.0` in order:
- #65054
- #65147
- #65332
- #65354
- #65094
This includes task executor scan handle validation, S3 TVF regression
isolation, Paimon JNI IOManager/profile improvements, and mixed-case
external table column handling for Iceberg/Paimon.
## Notes
- PR #65094 was adapted for `branch-4.0`; the final JDBC catalog
regression expectation commit only touched a test file that is already
deleted on this branch, so it was skipped as empty after conflict
resolution.
- The Iceberg sort-order fragment from #65094 was not applied because
`SortFieldInfo` and the corresponding create-table sort-order path are
not present on `branch-4.0`.
## Validation
- `git diff --check origin/branch-4.0..HEAD`
Cherry-picked from apache/doris#65054, apache/doris#65147,
apache/doris#65332, apache/doris#65354, and apache/doris#65094.
---
.../time_sharing/time_sharing_task_executor.cpp | 28 +-
be/src/vec/exec/format/table/paimon_jni_reader.cpp | 24 ++
be/src/vec/exec/format/table/paimon_jni_reader.h | 2 +
be/src/vec/exec/jni_connector.cpp | 36 ++-
be/src/vec/exec/scan/scanner_scheduler.h | 22 +-
.../time_sharing_task_executor_test.cpp | 83 ++++-
.../org/apache/doris/paimon/PaimonJniScanner.java | 359 ++++++++++++++++++++-
.../doris/paimon/PaimonSysTableJniScanner.java | 8 +-
.../apache/doris/paimon/PaimonJniScannerTest.java | 243 ++++++++++++++
.../datasource/iceberg/DorisTypeToIcebergType.java | 13 +-
.../datasource/iceberg/IcebergMetadataOps.java | 4 +-
.../doris/datasource/iceberg/IcebergUtils.java | 26 +-
.../datasource/paimon/PaimonExternalTable.java | 2 +-
.../apache/doris/datasource/paimon/PaimonUtil.java | 2 +-
.../paimon/source/PaimonPredicateConverter.java | 15 +-
.../datasource/paimon/source/PaimonScanNode.java | 81 ++++-
.../tablefunction/PaimonTableValuedFunction.java | 35 +-
.../datasource/iceberg/CreateIcebergTableTest.java | 22 ++
.../doris/datasource/iceberg/IcebergUtilsTest.java | 13 +
.../doris/datasource/paimon/PaimonUtilTest.java | 40 +++
.../paimon/source/PaimonScanNodeTest.java | 65 ++++
.../iceberg/test_iceberg_invaild_avro_name.out | 5 +-
.../paimon/test_paimon_catalog.groovy | 3 +-
23 files changed, 1071 insertions(+), 60 deletions(-)
diff --git
a/be/src/vec/exec/executor/time_sharing/time_sharing_task_executor.cpp
b/be/src/vec/exec/executor/time_sharing/time_sharing_task_executor.cpp
index 4459f20037b..876645927fa 100644
--- a/be/src/vec/exec/executor/time_sharing/time_sharing_task_executor.cpp
+++ b/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();
diff --git a/be/src/vec/exec/format/table/paimon_jni_reader.cpp
b/be/src/vec/exec/format/table/paimon_jni_reader.cpp
index 3c9afe93eb3..c8cd4b30910 100644
--- a/be/src/vec/exec/format/table/paimon_jni_reader.cpp
+++ b/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())
{
diff --git a/be/src/vec/exec/format/table/paimon_jni_reader.h
b/be/src/vec/exec/format/table/paimon_jni_reader.h
index 81b5bd68d29..78c17f9f2db 100644
--- a/be/src/vec/exec/format/table/paimon_jni_reader.h
+++ b/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);
diff --git a/be/src/vec/exec/jni_connector.cpp
b/be/src/vec/exec/jni_connector.cpp
index 2cad5967e86..df7f4f9e0ec 100644
--- a/be/src/vec/exec/jni_connector.cpp
+++ b/be/src/vec/exec/jni_connector.cpp
@@ -869,6 +869,9 @@ 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) {
@@ -876,22 +879,49 @@ void JniConnector::_collect_profile_before_close() {
<< "'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);
}
}
}
diff --git a/be/src/vec/exec/scan/scanner_scheduler.h
b/be/src/vec/exec/scan/scanner_scheduler.h
index 089f3e1e5b7..1790c30faa8 100644
--- a/be/src/vec/exec/scan/scanner_scheduler.h
+++ b/be/src/vec/exec/scan/scanner_scheduler.h
@@ -294,13 +294,28 @@ public:
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 @@ public:
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();
diff --git
a/be/test/vec/exec/executor/time_sharing/time_sharing_task_executor_test.cpp
b/be/test/vec/exec/executor/time_sharing/time_sharing_task_executor_test.cpp
index a1e604cb6f9..70187f9ef66 100644
--- a/be/test/vec/exec/executor/time_sharing/time_sharing_task_executor_test.cpp
+++ b/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 @@ private:
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 @@ private:
}
};
+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>();
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 5690b7f6505..cfbed4969da 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
@@ -25,26 +25,49 @@ import
org.apache.doris.common.security.authentication.PreExecutionAuthenticator
import com.google.common.base.Preconditions;
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.management.ManagementFactory;
+import java.lang.management.MemoryMXBean;
+import java.lang.management.MemoryUsage;
+import java.lang.management.ThreadInfo;
+import java.lang.management.ThreadMXBean;
+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.HashMap;
import java.util.List;
+import java.util.Locale;
import java.util.Map;
+import java.util.Optional;
import java.util.TimeZone;
+import java.util.concurrent.atomic.AtomicInteger;
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.";
+ 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";
+ 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 static final AtomicInteger ACTIVE_SCANNERS = new AtomicInteger();
private final Map<String, String> params;
private final Map<String, String> hadoopOptionParams;
@@ -52,12 +75,20 @@ 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;
private RecordReader.RecordIterator<InternalRow> recordIterator = null;
private final ClassLoader classLoader;
private PreExecutionAuthenticator preExecutionAuthenticator;
+ private boolean scannerCounted;
+ private long openTimeNanos;
+ private long readBatchTimeNanos;
+ private long readBatchCalls;
+ private long emptyReadBatchCalls;
+ private long rowsRead;
public PaimonJniScanner(int batchSize, Map<String, String> params) {
this.classLoader = this.getClass().getClassLoader();
@@ -65,8 +96,11 @@ public class PaimonJniScanner extends JniScanner {
LOG.debug("params:{}", params);
}
this.params = params;
- String[] requiredFields = params.get("required_fields").split(",");
- String[] requiredTypes = params.get("columns_types").split("#");
+ String[] requiredFields =
splitRequiredParam(params.get("required_fields"), ",");
+ String[] requiredTypes =
splitRequiredParam(params.get("columns_types"), "#");
+ Preconditions.checkArgument(requiredFields.length ==
requiredTypes.length,
+ "Required fields size %s is not matched with required types
size %s",
+ requiredFields.length, requiredTypes.length);
ColumnType[] columnTypes = new ColumnType[requiredTypes.length];
for (int i = 0; i < requiredTypes.length; i++) {
columnTypes[i] = ColumnType.parseType(requiredFields[i],
requiredTypes[i]);
@@ -85,6 +119,8 @@ public class PaimonJniScanner extends JniScanner {
@Override
public void open() throws IOException {
+ markScannerOpenedForMetrics();
+ long startTime = System.nanoTime();
try {
// When the user does not specify hive-site.xml, Paimon will look
for the file from the classpath:
// org.apache.paimon.hive.HiveCatalog.createHiveConf:
@@ -99,8 +135,15 @@ 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);
+ } finally {
+ openTimeNanos += System.nanoTime() - startTime;
}
}
@@ -116,13 +159,118 @@ 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(paimonAllFieldNames::indexOf).toArray();
+ return Arrays.stream(fields).mapToInt(fieldName -> {
+ int index = getFieldIndex(paimonAllFieldNames, fieldName);
+ Preconditions.checkArgument(index >= 0, "RequiredField %s not
found in schema", fieldName);
+ return index;
+ }).toArray();
+ }
+
+ static int getFieldIndex(List<String> fieldNames, String fieldName) {
+ for (int i = 0; i < fieldNames.size(); i++) {
+ if (fieldNames.get(i).equalsIgnoreCase(fieldName)) {
+ return i;
+ }
+ }
+ return -1;
+ }
+
+ static String[] splitRequiredParam(String value, String delimiterRegex) {
+ if (value == null || value.isEmpty()) {
+ return new String[0];
+ }
+ return value.split(delimiterRegex);
}
private List<Predicate> getPredicates() {
@@ -146,7 +294,7 @@ public class PaimonJniScanner extends JniScanner {
if (types[i].isDateTimeV2()) {
// paimon support precision > 6, but it has been reset as 6 in
FE
// try to get the right precision for datetimev2
- int index = paimonAllFieldNames.indexOf(fields[i]);
+ int index = getFieldIndex(paimonAllFieldNames, fields[i]);
if (index != -1) {
DataType dataType = table.rowType().getTypeAt(index);
if (dataType instanceof TimestampType) {
@@ -159,8 +307,36 @@ public class PaimonJniScanner extends JniScanner {
@Override
public void close() throws IOException {
- if (reader != null) {
- reader.close();
+ IOException exception = null;
+ try {
+ if (reader != null) {
+ 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;
+ }
+ }
+ } finally {
+ markScannerClosedForMetrics();
+ }
+ if (exception != null) {
+ throw exception;
}
}
@@ -168,7 +344,7 @@ public class PaimonJniScanner extends JniScanner {
int rows = 0;
try {
if (recordIterator == null) {
- recordIterator = reader.readBatch();
+ recordIterator = readBatchWithMetrics();
}
while (recordIterator != null) {
@@ -183,15 +359,23 @@ public class PaimonJniScanner extends JniScanner {
appendData(i, columnValue);
}
if (rows >= batchSize) {
+ if (fields.length == 0) {
+ vectorTable.appendVirtualData(rows);
+ }
appendDataTime += System.nanoTime() - startTime;
+ rowsRead += rows;
return rows;
}
}
appendDataTime += System.nanoTime() - startTime;
recordIterator.releaseBatch();
- recordIterator = reader.readBatch();
+ recordIterator = readBatchWithMetrics();
}
+ if (fields.length == 0 && rows > 0) {
+ vectorTable.appendVirtualData(rows);
+ }
+ rowsRead += rows;
} catch (Exception e) {
close();
LOG.warn("Failed to get the next batch of paimon. "
@@ -202,6 +386,20 @@ public class PaimonJniScanner extends JniScanner {
return rows;
}
+ private RecordReader.RecordIterator<InternalRow> readBatchWithMetrics()
throws IOException {
+ long startTime = System.nanoTime();
+ try {
+ RecordReader.RecordIterator<InternalRow> iterator =
reader.readBatch();
+ if (iterator == null) {
+ emptyReadBatchCalls++;
+ }
+ return iterator;
+ } finally {
+ readBatchCalls++;
+ readBatchTimeNanos += System.nanoTime() - startTime;
+ }
+ }
+
@Override
protected int getNext() {
try {
@@ -217,6 +415,148 @@ public class PaimonJniScanner extends JniScanner {
return null;
}
+ @Override
+ public Map<String, String> getStatistics() {
+ Map<String, String> statistics = new HashMap<>();
+ statistics.put("gauge:PaimonJniIOManagerEnabled", ioManager != null ?
"1" : "0");
+ statistics.put("gauge:PaimonJniActiveScannerCount",
String.valueOf(ACTIVE_SCANNERS.get()));
+ statistics.put("gauge:PaimonJniAsyncReaderThreadCount",
+ String.valueOf(currentAsyncReaderThreadCount()));
+ statistics.put("gauge:PaimonJniRequiredFieldCount",
String.valueOf(fields.length));
+ statistics.put("counter:PaimonJniSplitEncodedLength",
String.valueOf(lengthOfParam("paimon_split")));
+ statistics.put("counter:PaimonJniPredicateEncodedLength",
String.valueOf(lengthOfParam("paimon_predicate")));
+ statistics.put("gauge:PaimonJniAsyncThresholdConfigured",
+ hasPaimonOption(FILE_READER_ASYNC_THRESHOLD) ? "1" : "0");
+
parseDataSizeBytes(paimonOption(FILE_READER_ASYNC_THRESHOLD)).ifPresent(
+ bytes ->
statistics.put("bytes_gauge:PaimonJniAsyncThresholdBytes",
String.valueOf(bytes)));
+ statistics.put("counter:PaimonJniReadBatchCalls",
String.valueOf(readBatchCalls));
+ statistics.put("counter:PaimonJniEmptyReadBatchCalls",
String.valueOf(emptyReadBatchCalls));
+ statistics.put("counter:PaimonJniRowsRead", String.valueOf(rowsRead));
+ statistics.put("timer:PaimonJniScannerOpenTime",
String.valueOf(openTimeNanos));
+ statistics.put("timer:PaimonJniReadBatchTime",
String.valueOf(readBatchTimeNanos));
+ putMemoryStatistics(statistics);
+ return statistics;
+ }
+
+ private int lengthOfParam(String key) {
+ String value = params.get(key);
+ return value == null ? 0 : value.length();
+ }
+
+ private boolean hasPaimonOption(String key) {
+ return paimonOption(key) != null;
+ }
+
+ private String paimonOption(String key) {
+ if (table != null) {
+ String tableOption = table.options().get(key);
+ if (tableOption != null) {
+ return tableOption;
+ }
+ }
+ return params.get(PAIMON_OPTION_PREFIX + key);
+ }
+
+ private static void putMemoryStatistics(Map<String, String> statistics) {
+ MemoryMXBean memoryMXBean = ManagementFactory.getMemoryMXBean();
+ MemoryUsage heapUsage = memoryMXBean.getHeapMemoryUsage();
+ MemoryUsage nonHeapUsage = memoryMXBean.getNonHeapMemoryUsage();
+ statistics.put("bytes_gauge:PaimonJniJvmHeapUsed",
String.valueOf(nonNegative(heapUsage.getUsed())));
+ statistics.put("bytes_gauge:PaimonJniJvmHeapCommitted",
String.valueOf(nonNegative(heapUsage.getCommitted())));
+ statistics.put("bytes_gauge:PaimonJniJvmHeapMax",
String.valueOf(nonNegative(heapUsage.getMax())));
+ statistics.put("bytes_gauge:PaimonJniJvmNonHeapUsed",
String.valueOf(nonNegative(nonHeapUsage.getUsed())));
+ statistics.put("bytes_gauge:PaimonJniJvmNonHeapCommitted",
+ String.valueOf(nonNegative(nonHeapUsage.getCommitted())));
+ statistics.put("bytes_gauge:PaimonJniJvmNonHeapMax",
String.valueOf(nonNegative(nonHeapUsage.getMax())));
+ }
+
+ private static long nonNegative(long value) {
+ return Math.max(value, 0L);
+ }
+
+ private static int currentAsyncReaderThreadCount() {
+ return countThreadsByNamePrefix(ASYNC_READER_THREAD_NAME_PREFIX);
+ }
+
+ static int countThreadsByNamePrefix(String threadNamePrefix) {
+ int count = 0;
+ ThreadMXBean threadMXBean = ManagementFactory.getThreadMXBean();
+ ThreadInfo[] threadInfos =
threadMXBean.getThreadInfo(threadMXBean.getAllThreadIds(), 0);
+ for (ThreadInfo threadInfo : threadInfos) {
+ if (threadInfo != null &&
threadInfo.getThreadName().startsWith(threadNamePrefix)) {
+ count++;
+ }
+ }
+ return count;
+ }
+
+ private void markScannerOpenedForMetrics() {
+ if (!scannerCounted) {
+ scannerCounted = true;
+ ACTIVE_SCANNERS.incrementAndGet();
+ }
+ }
+
+ private void markScannerClosedForMetrics() {
+ if (scannerCounted) {
+ scannerCounted = false;
+ ACTIVE_SCANNERS.decrementAndGet();
+ }
+ }
+
+ static Optional<Long> parseDataSizeBytes(String value) {
+ if (value == null || value.trim().isEmpty()) {
+ return Optional.empty();
+ }
+ String normalized = value.trim().toLowerCase(Locale.ROOT).replace("_",
"").replace(" ", "");
+ int unitStart = 0;
+ while (unitStart < normalized.length()
+ && (Character.isDigit(normalized.charAt(unitStart)) ||
normalized.charAt(unitStart) == '.')) {
+ unitStart++;
+ }
+ if (unitStart == 0) {
+ return Optional.empty();
+ }
+ try {
+ double number = Double.parseDouble(normalized.substring(0,
unitStart));
+ String unit = normalized.substring(unitStart);
+ long multiplier;
+ switch (unit) {
+ case "":
+ case "b":
+ case "byte":
+ case "bytes":
+ multiplier = 1L;
+ break;
+ case "k":
+ case "kb":
+ case "kib":
+ multiplier = 1024L;
+ break;
+ case "m":
+ case "mb":
+ case "mib":
+ multiplier = 1024L * 1024L;
+ break;
+ case "g":
+ case "gb":
+ case "gib":
+ multiplier = 1024L * 1024L * 1024L;
+ break;
+ case "t":
+ case "tb":
+ case "tib":
+ multiplier = 1024L * 1024L * 1024L * 1024L;
+ break;
+ default:
+ return Optional.empty();
+ }
+ return Optional.of((long) (number * multiplier));
+ } catch (NumberFormatException e) {
+ return Optional.empty();
+ }
+ }
+
private void initTable() {
Preconditions.checkState(params.containsKey("serialized_table"));
table = PaimonUtils.deserialize(params.get("serialized_table"));
@@ -227,4 +567,3 @@ public class PaimonJniScanner extends JniScanner {
}
}
-
diff --git
a/fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonSysTableJniScanner.java
b/fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonSysTableJniScanner.java
index 6c2eab84129..90d309defcc 100644
---
a/fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonSysTableJniScanner.java
+++
b/fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonSysTableJniScanner.java
@@ -85,7 +85,11 @@ public class PaimonSysTableJniScanner extends JniScanner {
LOG.debug("paimonAllFieldNames:{}", paimonAllFieldNames);
}
resetDatetimeV2Precision();
- this.projected =
Arrays.stream(fields).mapToInt(paimonAllFieldNames::indexOf).toArray();
+ this.projected = Arrays.stream(fields).mapToInt(fieldName -> {
+ int index = PaimonJniScanner.getFieldIndex(paimonAllFieldNames,
fieldName);
+ Preconditions.checkArgument(index >= 0, "RequiredField %s not
found in schema", fieldName);
+ return index;
+ }).toArray();
this.paimonDataTypeList = Arrays.stream(projected).mapToObj(i ->
table.rowType().getTypeAt(i))
.collect(Collectors.toList());
this.paimonSplits =
Arrays.stream(params.get("serialized_splits").split(","))
@@ -148,7 +152,7 @@ public class PaimonSysTableJniScanner extends JniScanner {
if (types[i].isDateTimeV2()) {
// paimon support precision > 6, but it has been reset as 6 in
FE
// try to get the right precision for datetimev2
- int index = paimonAllFieldNames.indexOf(fields[i]);
+ int index =
PaimonJniScanner.getFieldIndex(paimonAllFieldNames, fields[i]);
if (index != -1) {
DataType dataType = table.rowType().getTypeAt(index);
if (dataType instanceof TimestampType) {
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
new file mode 100644
index 00000000000..7ab07c8efcf
--- /dev/null
+++
b/fe/be-java-extensions/paimon-scanner/src/test/java/org/apache/doris/paimon/PaimonJniScannerTest.java
@@ -0,0 +1,243 @@
+// 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.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.apache.paimon.table.Table;
+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.lang.reflect.Proxy;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+public class PaimonJniScannerTest {
+ @Rule
+ public TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+ @Test
+ public void testConstructorAcceptsEmptyProjection() {
+ new PaimonJniScanner(128, createBaseParams());
+ Assert.assertArrayEquals(new String[0],
PaimonJniScanner.splitRequiredParam("", ","));
+ Assert.assertArrayEquals(new String[0],
PaimonJniScanner.splitRequiredParam("", "#"));
+ Assert.assertArrayEquals(new String[] {"id", "name"},
+ PaimonJniScanner.splitRequiredParam("id,name", ","));
+ }
+
+ @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("gauge:PaimonJniIOManagerEnabled"));
+
+ scanner.close();
+ Assert.assertFalse(spillDir.exists());
+ }
+
+ @Test
+ public void testStatisticsIncludePaimonDiagnostics() throws Exception {
+ Map<String, String> params = createBaseParams();
+ params.put("paimon_split", "encoded-split");
+ params.put("paimon_predicate", "encoded-predicate");
+ PaimonJniScanner scanner = new PaimonJniScanner(128, params);
+ setTableOptions(scanner,
Collections.singletonMap("file-reader-async-threshold", "10 MiB"));
+
+ Map<String, String> statistics = scanner.getStatistics();
+
+ Assert.assertEquals("0",
statistics.get("gauge:PaimonJniIOManagerEnabled"));
+ Assert.assertEquals("0",
statistics.get("gauge:PaimonJniRequiredFieldCount"));
+ Assert.assertEquals("13",
statistics.get("counter:PaimonJniSplitEncodedLength"));
+ Assert.assertEquals("17",
statistics.get("counter:PaimonJniPredicateEncodedLength"));
+ Assert.assertEquals("1",
statistics.get("gauge:PaimonJniAsyncThresholdConfigured"));
+ Assert.assertEquals(String.valueOf(10L * 1024L * 1024L),
+ statistics.get("bytes_gauge:PaimonJniAsyncThresholdBytes"));
+
Assert.assertTrue(statistics.containsKey("gauge:PaimonJniAsyncReaderThreadCount"));
+
Assert.assertTrue(statistics.containsKey("gauge:PaimonJniActiveScannerCount"));
+
Assert.assertFalse(statistics.containsKey("peak:PaimonJniActiveScannerPeakCount"));
+
Assert.assertFalse(statistics.containsKey("peak:PaimonJniAsyncReaderThreadPeakCount"));
+
Assert.assertTrue(statistics.containsKey("counter:PaimonJniReadBatchCalls"));
+
Assert.assertTrue(statistics.containsKey("timer:PaimonJniScannerOpenTime"));
+
Assert.assertTrue(statistics.containsKey("timer:PaimonJniReadBatchTime"));
+
Assert.assertTrue(Long.parseLong(statistics.get("bytes_gauge:PaimonJniJvmHeapUsed"))
> 0);
+
Assert.assertTrue(Long.parseLong(statistics.get("bytes_gauge:PaimonJniJvmHeapCommitted"))
> 0);
+ }
+
+ @Test
+ public void testCountThreadsByNamePrefix() throws Exception {
+ CountDownLatch started = new CountDownLatch(1);
+ CountDownLatch release = new CountDownLatch(1);
+ Thread thread = new Thread(() -> {
+ started.countDown();
+ try {
+ release.await(30, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }, "paimon-reader-async-thread-test");
+
+ thread.start();
+ try {
+ Assert.assertTrue(started.await(5, TimeUnit.SECONDS));
+
Assert.assertTrue(PaimonJniScanner.countThreadsByNamePrefix("paimon-reader-async-thread")
>= 1);
+ } finally {
+ release.countDown();
+ thread.join(5000);
+ }
+ }
+
+ @Test
+ public void testParseDataSizeBytes() {
+ Assert.assertEquals(Long.valueOf(1024L),
PaimonJniScanner.parseDataSizeBytes("1 KiB").get());
+ Assert.assertEquals(Long.valueOf(10L * 1024L * 1024L),
+ PaimonJniScanner.parseDataSizeBytes("10 MiB").get());
+ Assert.assertEquals(Long.valueOf(2L * 1024L * 1024L * 1024L),
+ PaimonJniScanner.parseDataSizeBytes("2GB").get());
+
Assert.assertFalse(PaimonJniScanner.parseDataSizeBytes("unknown").isPresent());
+ }
+
+ 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;
+ }
+
+ private void setTableOptions(PaimonJniScanner scanner, Map<String, String>
options) throws Exception {
+ Table table = (Table) Proxy.newProxyInstance(
+ Table.class.getClassLoader(), new Class[] {Table.class},
(proxy, method, args) -> {
+ if ("options".equals(method.getName())) {
+ return options;
+ }
+ if ("toString".equals(method.getName())) {
+ return "TestPaimonTable";
+ }
+ throw new UnsupportedOperationException(method.getName());
+ });
+ Field tableField = PaimonJniScanner.class.getDeclaredField("table");
+ tableField.setAccessible(true);
+ tableField.set(scanner, table);
+ }
+
+ 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();
+ }
+
+ @Override
+ public void close() {
+ }
+ }
+
+ @Test
+ public void testGetFieldIndexMatchesMixedCaseColumns() {
+ Assert.assertEquals(1,
PaimonJniScanner.getFieldIndex(Arrays.asList("data", "mIxEd_COL", "PART"),
+ "mixed_col"));
+ Assert.assertEquals(2,
PaimonJniScanner.getFieldIndex(Arrays.asList("data", "mIxEd_COL", "PART"),
+ "part"));
+ Assert.assertEquals(-1,
PaimonJniScanner.getFieldIndex(Arrays.asList("data", "mIxEd_COL", "PART"),
+ "missing_col"));
+ }
+}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/DorisTypeToIcebergType.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/DorisTypeToIcebergType.java
index 56fa03120d9..de19d90728d 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/DorisTypeToIcebergType.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/DorisTypeToIcebergType.java
@@ -29,6 +29,7 @@ import com.google.common.collect.Lists;
import org.apache.iceberg.types.Type;
import org.apache.iceberg.types.Types;
+import java.util.Collections;
import java.util.List;
@@ -37,14 +38,21 @@ import java.util.List;
*/
public class DorisTypeToIcebergType extends DorisTypeVisitor<Type> {
private final StructType root;
+ private final List<String> rootFieldNames;
private int nextId = 0;
public DorisTypeToIcebergType() {
this.root = null;
+ this.rootFieldNames = Collections.emptyList();
}
public DorisTypeToIcebergType(StructType root) {
+ this(root, Collections.emptyList());
+ }
+
+ public DorisTypeToIcebergType(StructType root, List<String>
rootFieldNames) {
this.root = root;
+ this.rootFieldNames = rootFieldNames;
// the root struct's fields use the first ids
this.nextId = root.getFields().size();
}
@@ -65,10 +73,11 @@ public class DorisTypeToIcebergType extends
DorisTypeVisitor<Type> {
Type type = types.get(i);
int id = isRoot ? i : getNextId();
+ String fieldName = isRoot && !rootFieldNames.isEmpty() ?
rootFieldNames.get(i) : field.getName();
if (field.getContainsNull()) {
- newFields.add(Types.NestedField.optional(id, field.getName(),
type, field.getComment()));
+ newFields.add(Types.NestedField.optional(id, fieldName, type,
field.getComment()));
} else {
- newFields.add(Types.NestedField.required(id, field.getName(),
type, field.getComment()));
+ newFields.add(Types.NestedField.required(id, fieldName, type,
field.getComment()));
}
}
return Types.StructType.of(newFields);
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergMetadataOps.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergMetadataOps.java
index 545237cea27..3aa26f3572c 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergMetadataOps.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergMetadataOps.java
@@ -353,8 +353,8 @@ public class IcebergMetadataOps implements
ExternalMetadataOps {
.map(col -> new StructField(col.getName(), col.getType(),
col.getComment(), col.isAllowNull()))
.collect(Collectors.toList());
StructType structType = new StructType(new ArrayList<>(collect));
- Type visit =
- DorisTypeVisitor.visit(structType, new
DorisTypeToIcebergType(structType));
+ List<String> rootFieldNames =
columns.stream().map(Column::getName).collect(Collectors.toList());
+ Type visit = DorisTypeVisitor.visit(structType, new
DorisTypeToIcebergType(structType, rootFieldNames));
Schema schema = new
Schema(visit.asNestedType().asStructType().fields());
Map<String, String> properties = createTableInfo.getProperties();
properties.put(ExternalCatalog.DORIS_VERSION,
ExternalCatalog.DORIS_VERSION_VALUE);
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergUtils.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergUtils.java
index 9895ceb7023..222809ca6a4 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergUtils.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergUtils.java
@@ -133,7 +133,6 @@ import java.util.ArrayList;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
-import java.util.Locale;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
@@ -520,34 +519,38 @@ public class IcebergUtils {
PartitionSpec.Builder builder = PartitionSpec.builderFor(schema);
for (Expr expr : partitionExprs) {
if (expr instanceof SlotRef) {
- builder.identity(((SlotRef) expr).getColumnName());
+ builder.identity(getIcebergColumnName(schema, ((SlotRef)
expr).getColumnName()));
} else if (expr instanceof FunctionCallExpr) {
String exprName = expr.getExprName();
List<Expr> params = ((FunctionCallExpr)
expr).getParams().exprs();
switch (exprName.toLowerCase()) {
case "bucket":
- builder.bucket(params.get(1).getExprName(),
Integer.parseInt(params.get(0).getStringValue()));
+ builder.bucket(
+ getIcebergColumnName(schema,
params.get(1).getExprName()),
+
Integer.parseInt(params.get(0).getStringValue()));
break;
case "year":
case "years":
- builder.year(params.get(0).getExprName());
+ builder.year(getIcebergColumnName(schema,
params.get(0).getExprName()));
break;
case "month":
case "months":
- builder.month(params.get(0).getExprName());
+ builder.month(getIcebergColumnName(schema,
params.get(0).getExprName()));
break;
case "date":
case "day":
case "days":
- builder.day(params.get(0).getExprName());
+ builder.day(getIcebergColumnName(schema,
params.get(0).getExprName()));
break;
case "date_hour":
case "hour":
case "hours":
- builder.hour(params.get(0).getExprName());
+ builder.hour(getIcebergColumnName(schema,
params.get(0).getExprName()));
break;
case "truncate":
- builder.truncate(params.get(1).getExprName(),
Integer.parseInt(params.get(0).getStringValue()));
+ builder.truncate(
+ getIcebergColumnName(schema,
params.get(1).getExprName()),
+
Integer.parseInt(params.get(0).getStringValue()));
break;
default:
throw new UserException("unsupported partition for " +
exprName);
@@ -557,6 +560,11 @@ public class IcebergUtils {
return builder.build();
}
+ private static String getIcebergColumnName(Schema schema, String
columnName) {
+ Types.NestedField field = schema.caseInsensitiveFindField(columnName);
+ return field == null ? columnName : field.name();
+ }
+
private static Type
icebergPrimitiveTypeToDorisType(org.apache.iceberg.types.Type.PrimitiveType
primitive,
boolean enableMappingVarbinary, boolean enableMappingTimestampTz) {
switch (primitive.typeId()) {
@@ -974,7 +982,7 @@ public class IcebergUtils {
List<Types.NestedField> columns = schema.columns();
List<Column> resSchema =
Lists.newArrayListWithCapacity(columns.size());
for (Types.NestedField field : columns) {
- Column column = new Column(field.name().toLowerCase(Locale.ROOT),
+ Column column = new Column(field.name(),
IcebergUtils.icebergTypeToDorisType(field.type(),
enableMappingVarbinary, enableMappingTimestampTz),
true, null,
true, field.doc(), true, -1);
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonExternalTable.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonExternalTable.java
index 1782bae59b0..1a99aad60f6 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonExternalTable.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonExternalTable.java
@@ -330,7 +330,7 @@ public class PaimonExternalTable extends ExternalTable
implements MTMVRelatedTab
Set<String> partitionColumnNames =
Sets.newHashSet(tableSchema.partitionKeys());
List<Column> partitionColumns = Lists.newArrayList();
for (DataField field : columns) {
- Column column = new Column(field.name().toLowerCase(),
+ Column column = new Column(field.name(),
PaimonUtil.paimonTypeToDorisType(field.type(),
getCatalog().getEnableMappingVarbinary(),
getCatalog().getEnableMappingTimestampTz()),
true,
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonUtil.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonUtil.java
index 08358a3da99..3239035f20d 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonUtil.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonUtil.java
@@ -430,7 +430,7 @@ public class PaimonUtil {
boolean enableTimestampTzMapping) {
List<Column> resSchema =
Lists.newArrayListWithCapacity(rowType.getFields().size());
rowType.getFields().forEach(field -> {
- resSchema.add(new Column(field.name().toLowerCase(),
+ resSchema.add(new Column(field.name(),
PaimonUtil.paimonTypeToDorisType(field.type(),
enableVarbinaryMapping, enableTimestampTzMapping),
primaryKeys.contains(field.name()),
null,
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonPredicateConverter.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonPredicateConverter.java
index 73a3c72ddcc..ae45c242718 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonPredicateConverter.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonPredicateConverter.java
@@ -46,7 +46,7 @@ public class PaimonPredicateConverter {
public PaimonPredicateConverter(RowType rowType) {
this.builder = new PredicateBuilder(rowType);
- this.fieldNames = rowType.getFields().stream().map(f ->
f.name().toLowerCase()).collect(Collectors.toList());
+ this.fieldNames =
rowType.getFields().stream().map(DataField::name).collect(Collectors.toList());
this.paimonFieldTypes =
rowType.getFields().stream().map(DataField::type).collect(Collectors.toList());
}
@@ -99,7 +99,7 @@ public class PaimonPredicateConverter {
return null;
}
String colName = slotRef.getColumnName();
- int idx = fieldNames.indexOf(colName);
+ int idx = getFieldIndex(colName);
DataType dataType = paimonFieldTypes.get(idx);
List<Object> valueList = new ArrayList<>();
for (int i = 1; i < predicate.getChildren().size(); i++) {
@@ -132,7 +132,7 @@ public class PaimonPredicateConverter {
return null;
}
String colName = slotRef.getColumnName();
- int idx = fieldNames.indexOf(colName);
+ int idx = getFieldIndex(colName);
DataType dataType = paimonFieldTypes.get(idx);
Object value = dataType.accept(new PaimonValueConverter(literalExpr));
if (value == null) {
@@ -174,6 +174,15 @@ public class PaimonPredicateConverter {
}
+ private int getFieldIndex(String colName) {
+ for (int i = 0; i < fieldNames.size(); i++) {
+ if (fieldNames.get(i).equalsIgnoreCase(colName)) {
+ return i;
+ }
+ }
+ return fieldNames.indexOf(colName);
+ }
+
public static SlotRef convertDorisExprToSlotRef(Expr expr) {
SlotRef slotRef = null;
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 00359b5a11c..5b24ddfccde 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
@@ -19,6 +19,7 @@ package org.apache.doris.datasource.paimon.source;
import org.apache.doris.analysis.TableScanParams;
import org.apache.doris.analysis.TupleDescriptor;
+import org.apache.doris.catalog.JdbcResource;
import org.apache.doris.catalog.TableIf;
import org.apache.doris.common.DdlException;
import org.apache.doris.common.MetaNotFoundException;
@@ -65,6 +66,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;
@@ -89,6 +91,17 @@ 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 String JDBC_PREFIX = "jdbc.";
+ private static final String JDBC_DRIVER_URL = JDBC_PREFIX +
JdbcResource.DRIVER_URL;
+ private static final String JDBC_DRIVER_CLASS = JDBC_PREFIX +
JdbcResource.DRIVER_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 enum SplitReadType {
JNI,
@@ -142,6 +155,7 @@ public class PaimonScanNode extends FileQueryScanNode {
// get them in doInitialize() to ensure internal consistency of ScanNode
private Map<StorageProperties.Type, StorageProperties>
storagePropertiesMap;
private Map<String, String> backendStorageProperties;
+ private Map<String, String> backendPaimonOptions = Collections.emptyMap();
// The schema information involved in the current query process (including
historical schema).
protected ConcurrentHashMap<Long, Boolean> currentQuerySchema = new
ConcurrentHashMap<>();
@@ -169,6 +183,7 @@ public class PaimonScanNode extends FileQueryScanNode {
source.getPaimonTable()
);
backendStorageProperties =
CredentialUtils.getBackendPropertiesFromStorageMap(storagePropertiesMap);
+ backendPaimonOptions = getBackendPaimonOptions();
}
@VisibleForTesting
@@ -259,6 +274,9 @@ public class PaimonScanNode extends FileQueryScanNode {
// MUST explicitly set to -1, to be distinct from valid row count
>= 0
tableFormatFileDesc.setTableLevelRowCount(-1);
}
+ if (!backendPaimonOptions.isEmpty()) {
+ fileDesc.setPaimonOptions(backendPaimonOptions);
+ }
tableFormatFileDesc.setPaimonParams(fileDesc);
Map<String, String> partitionValues =
paimonSplit.getPaimonPartitionValues();
if (partitionValues != null) {
@@ -428,6 +446,51 @@ public class PaimonScanNode extends FileQueryScanNode {
return splits;
}
+ @VisibleForTesting
+ Map<String, String> getBackendPaimonOptions() {
+ if (source == null) {
+ return Collections.emptyMap();
+ }
+ if (!(source.getCatalog() instanceof PaimonExternalCatalog)) {
+ 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));
+ }
+ }
+ String driverUrl = getCatalogProperty(catalogProperties,
JdbcResource.DRIVER_URL);
+ if (driverUrl == null) {
+ return backendOptions;
+ }
+ String driverClass = getCatalogProperty(catalogProperties,
JdbcResource.DRIVER_CLASS);
+ if (driverClass == null) {
+ throw new IllegalArgumentException("jdbc.driver_class or
paimon.jdbc.driver_class is required when "
+ + "jdbc.driver_url or paimon.jdbc.driver_url is
specified");
+ }
+ backendOptions.put(JDBC_DRIVER_URL,
JdbcResource.getFullDriverUrl(driverUrl));
+ backendOptions.put(JDBC_DRIVER_CLASS, driverClass);
+ return backendOptions;
+ }
+
+ private String getCatalogProperty(Map<String, String> catalogProperties,
String property) {
+ String value = catalogProperties.get(PAIMON_PROPERTY_PREFIX +
JDBC_PREFIX + property);
+ if (value == null || value.trim().isEmpty()) {
+ value = catalogProperties.get(JDBC_PREFIX + property);
+ }
+ if (value == null || value.trim().isEmpty()) {
+ return null;
+ }
+ return value;
+ }
+
private long determineTargetFileSplitSize(List<DataSplit> dataSplits,
boolean isBatchMode) {
if (sessionVariable.getFileSplitSize() > 0) {
@@ -473,13 +536,9 @@ public class PaimonScanNode extends FileQueryScanNode {
@VisibleForTesting
public List<org.apache.paimon.table.source.Split> getPaimonSplitFromAPI()
throws UserException {
Table paimonTable = getProcessedTable();
+ List<String> fieldNames = paimonTable.rowType().getFieldNames();
int[] projected = desc.getSlots().stream().mapToInt(
- slot -> paimonTable.rowType()
- .getFieldNames()
- .stream()
- .map(String::toLowerCase)
- .collect(Collectors.toList())
- .indexOf(slot.getColumn().getName()))
+ slot -> getFieldIndex(fieldNames, slot.getColumn().getName()))
.filter(i -> i >= 0)
.toArray();
ReadBuilder readBuilder = paimonTable.newReadBuilder();
@@ -498,6 +557,16 @@ public class PaimonScanNode extends FileQueryScanNode {
return splits;
}
+ @VisibleForTesting
+ static int getFieldIndex(List<String> fieldNames, String columnName) {
+ for (int i = 0; i < fieldNames.size(); i++) {
+ if (fieldNames.get(i).equalsIgnoreCase(columnName)) {
+ return i;
+ }
+ }
+ return -1;
+ }
+
private String getFileFormat(String path) {
return
FileFormatUtils.getFileFormatBySuffix(path).orElse(source.getFileFormatFromTableProperties());
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/PaimonTableValuedFunction.java
b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/PaimonTableValuedFunction.java
index 525593cbd75..adca1635ef6 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/PaimonTableValuedFunction.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/PaimonTableValuedFunction.java
@@ -167,17 +167,18 @@ public class PaimonTableValuedFunction extends
MetadataTableValuedFunction {
@Override
public TMetaScanRange getMetaScanRange(List<String> requiredFileds) {
- int[] projections = requiredFileds.stream().mapToInt(
- field -> paimonSysTable.rowType().getFieldNames()
- .stream()
- .map(String::toLowerCase)
- .collect(Collectors.toList())
- .indexOf(field))
+ List<String> paimonFieldNames =
paimonSysTable.rowType().getFieldNames();
+ int[] projections = requiredFileds.stream()
+ .mapToInt(field -> getFieldIndex(paimonFieldNames, field))
.toArray();
List<Split> splits;
try {
- splits = hadoopAuthenticator.execute(
- () ->
paimonSysTable.newReadBuilder().withProjection(projections).newScan().plan().splits());
+ splits = hadoopAuthenticator.execute(() -> {
+ if (hasInvalidProjection(projections)) {
+ return
paimonSysTable.newReadBuilder().newScan().plan().splits();
+ }
+ return
paimonSysTable.newReadBuilder().withProjection(projections).newScan().plan().splits();
+ });
} catch (Exception e) {
throw new RuntimeException(ExceptionUtils.getRootCauseMessage(e));
}
@@ -191,6 +192,24 @@ public class PaimonTableValuedFunction extends
MetadataTableValuedFunction {
return tMetaScanRange;
}
+ private static boolean hasInvalidProjection(int[] projections) {
+ for (int projection : projections) {
+ if (projection < 0) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private static int getFieldIndex(List<String> fieldNames, String
fieldName) {
+ for (int i = 0; i < fieldNames.size(); i++) {
+ if (fieldNames.get(i).equalsIgnoreCase(fieldName)) {
+ return i;
+ }
+ }
+ return -1;
+ }
+
@Override
public String getTableName() {
return "PaimonTableValuedFunction<" + queryType + ">";
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/CreateIcebergTableTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/CreateIcebergTableTest.java
index 3422100de0f..0a23870c80f 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/CreateIcebergTableTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/CreateIcebergTableTest.java
@@ -177,6 +177,28 @@ public class CreateIcebergTableTest {
Assert.assertEquals("b", table.properties().get("a"));
}
+ @Test
+ public void testPartitionPreservesNonLowercaseColumnNames() throws
UserException {
+ TableIdentifier tb = TableIdentifier.of(dbName, getTableName());
+ String sql = "create table " + tb + " ("
+ + "data int, "
+ + "`PART` int, "
+ + "`mIxEd_COL` int"
+ + ") engine = iceberg "
+ + "partition by (`PART`, bucket(2, `mIxEd_COL`)) ()";
+ createTable(sql);
+ Table table = ops.getCatalog().loadTable(tb);
+ Schema schema = table.schema();
+
+ Assert.assertEquals("PART", schema.columns().get(1).name());
+ Assert.assertEquals("mIxEd_COL", schema.columns().get(2).name());
+ PartitionSpec spec = PartitionSpec.builderFor(schema)
+ .identity("PART")
+ .bucket("mIxEd_COL", 2)
+ .build();
+ Assert.assertEquals(spec, table.spec());
+ }
+
public void createTable(String sql) throws UserException {
LogicalPlan plan = new NereidsParser().parseSingle(sql);
Assertions.assertTrue(plan instanceof CreateTableCommand);
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergUtilsTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergUtilsTest.java
index e4ee0ec5e46..3f4828efb51 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergUtilsTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergUtilsTest.java
@@ -19,6 +19,7 @@ package org.apache.doris.datasource.iceberg;
import org.apache.doris.analysis.TableScanParams;
import org.apache.doris.analysis.TableSnapshot;
+import org.apache.doris.catalog.Column;
import org.apache.doris.common.UserException;
import org.apache.doris.datasource.iceberg.source.IcebergTableQueryInfo;
@@ -98,6 +99,18 @@ public class IcebergUtilsTest {
return declaredField.getBoolean(hiveCatalog);
}
+ @Test
+ public void testParseSchemaPreservesNonLowercaseColumnNames() {
+ Schema schema = new Schema(
+ Types.NestedField.required(1, "mIxEd_COL",
Types.IntegerType.get()),
+ Types.NestedField.required(2, "PART", Types.StringType.get()));
+
+ List<Column> columns = IcebergUtils.parseSchema(schema, false, false);
+
+ Assert.assertEquals("mIxEd_COL", columns.get(0).getName());
+ Assert.assertEquals("PART", columns.get(1).getName());
+ }
+
@Test
public void testGetMatchingManifest() {
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/PaimonUtilTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/PaimonUtilTest.java
index e06b7dee7bf..046b38e311b 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/PaimonUtilTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/PaimonUtilTest.java
@@ -17,13 +17,24 @@
package org.apache.doris.datasource.paimon;
+import org.apache.doris.catalog.Column;
import org.apache.doris.catalog.Type;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.table.Table;
import org.apache.paimon.types.CharType;
import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
import org.apache.paimon.types.VarCharType;
import org.junit.Assert;
import org.junit.Test;
+import org.mockito.Mockito;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
public class PaimonUtilTest {
@Test
@@ -36,4 +47,33 @@ public class PaimonUtilTest {
Assert.assertEquals(32, type1.getLength());
Assert.assertEquals(14, type2.getLength());
}
+
+ @Test
+ public void testParseSchemaPreservesNonLowercaseColumnNames() {
+ RowType rowType = DataTypes.ROW(
+ DataTypes.FIELD(0, "mIxEd_COL", DataTypes.INT()),
+ DataTypes.FIELD(1, "PART", DataTypes.STRING()));
+
+ List<Column> columns = PaimonUtil.parseSchema(rowType,
Collections.singletonList("PART"), false, false);
+
+ Assert.assertEquals("mIxEd_COL", columns.get(0).getName());
+ Assert.assertEquals("PART", columns.get(1).getName());
+ Assert.assertTrue(columns.get(1).isKey());
+ }
+
+ @Test
+ public void testGetPartitionInfoMapPreservesNonLowercaseKeys() {
+ DataField mixedCasePartition = DataTypes.FIELD(0, "Dt",
DataTypes.STRING());
+ Table table = Mockito.mock(Table.class);
+ Mockito.when(table.name()).thenReturn("mock_table");
+
Mockito.when(table.partitionKeys()).thenReturn(Collections.singletonList("Dt"));
+
Mockito.when(table.rowType()).thenReturn(DataTypes.ROW(mixedCasePartition));
+
+ BinaryRow partitionValues =
BinaryRow.singleColumn(BinaryString.fromString("2026-05-26"));
+
+ Map<String, String> partitionInfoMap =
PaimonUtil.getPartitionInfoMap(table, partitionValues, "UTC");
+
+ Assert.assertFalse(partitionInfoMap.containsKey("dt"));
+ Assert.assertEquals("2026-05-26", partitionInfoMap.get("Dt"));
+ }
}
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 454f2d42d9e..45d8c8909a4 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
@@ -21,8 +21,10 @@ import org.apache.doris.analysis.TupleDescriptor;
import org.apache.doris.analysis.TupleId;
import org.apache.doris.common.ExceptionChecker;
import org.apache.doris.common.UserException;
+import org.apache.doris.datasource.CatalogProperty;
import org.apache.doris.datasource.FileQueryScanNode;
import org.apache.doris.datasource.FileSplitter;
+import org.apache.doris.datasource.paimon.PaimonExternalCatalog;
import org.apache.doris.datasource.paimon.PaimonFileExternalCatalog;
import org.apache.doris.planner.PlanNodeId;
import org.apache.doris.planner.ScanContext;
@@ -43,6 +45,7 @@ import org.mockito.junit.MockitoJUnitRunner;
import java.lang.reflect.Method;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
@@ -408,6 +411,68 @@ public class PaimonScanNodeTest {
Assert.assertEquals(100L * 1024L * 1024L, target);
}
+ @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);
+
+ 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 = new PaimonScanNode(new PlanNodeId(0),
+ new TupleDescriptor(new TupleId(0)), false, sv,
ScanContext.EMPTY);
+ 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 testGetBackendPaimonOptionsForJdbcDriver() {
+ Map<String, String> props = new HashMap<>();
+ props.put("paimon.jdbc.driver_url", "file:///tmp/postgresql.jar");
+ props.put("paimon.jdbc.driver_class", "org.postgresql.Driver");
+
+ CatalogProperty catalogProperty = Mockito.mock(CatalogProperty.class);
+ Mockito.when(catalogProperty.getProperties()).thenReturn(props);
+
+ 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 = new PaimonScanNode(new PlanNodeId(0),
+ new TupleDescriptor(new TupleId(0)), false, sv,
ScanContext.EMPTY);
+ node.setSource(source);
+
+ Map<String, String> backendOptions = node.getBackendPaimonOptions();
+ Assert.assertEquals("file:///tmp/postgresql.jar",
backendOptions.get("jdbc.driver_url"));
+ Assert.assertEquals("org.postgresql.Driver",
backendOptions.get("jdbc.driver_class"));
+ Assert.assertEquals(2, backendOptions.size());
+ }
+
+ @Test
+ public void testGetFieldIndexMatchesMixedCaseColumns() {
+ List<String> fieldNames = Arrays.asList("data", "mIxEd_COL", "PART");
+
+ Assert.assertEquals(1, PaimonScanNode.getFieldIndex(fieldNames,
"mixed_col"));
+ Assert.assertEquals(2, PaimonScanNode.getFieldIndex(fieldNames,
"part"));
+ Assert.assertEquals(-1, PaimonScanNode.getFieldIndex(fieldNames,
"missing_col"));
+ }
+
private void mockJniReader(PaimonScanNode spyNode) {
Mockito.doReturn(false).when(spyNode).supportNativeReader(ArgumentMatchers.any(Optional.class));
}
diff --git
a/regression-test/data/external_table_p0/iceberg/test_iceberg_invaild_avro_name.out
b/regression-test/data/external_table_p0/iceberg/test_iceberg_invaild_avro_name.out
index 1b8af743c6a..be0c0e22364 100644
---
a/regression-test/data/external_table_p0/iceberg/test_iceberg_invaild_avro_name.out
+++
b/regression-test/data/external_table_p0/iceberg/test_iceberg_invaild_avro_name.out
@@ -1,7 +1,7 @@
-- This file is automatically generated. You should know what you did if you
want to edit this
-- !desc --
id int Yes true \N
-test:a1b2.raw.abc-gg-1-a text Yes true \N
+TEST:A1B2.RAW.ABC-GG-1-A text Yes true \N
-- !q_1 --
1 row1
@@ -29,7 +29,7 @@ test:a1b2.raw.abc-gg-1-a text Yes true \N
-- !desc --
id int Yes true \N
-test:a1b2.raw.abc-gg-1-a text Yes true \N
+TEST:A1B2.RAW.ABC-GG-1-A text Yes true \N
-- !q_1 --
1 row1
@@ -54,4 +54,3 @@ test:a1b2.raw.abc-gg-1-a text Yes true \N
3 row3
2 row2
1 row1
-
diff --git
a/regression-test/suites/external_table_p0/paimon/test_paimon_catalog.groovy
b/regression-test/suites/external_table_p0/paimon/test_paimon_catalog.groovy
index b5ca34e5a18..91d38cf7294 100644
--- a/regression-test/suites/external_table_p0/paimon/test_paimon_catalog.groovy
+++ b/regression-test/suites/external_table_p0/paimon/test_paimon_catalog.groovy
@@ -307,7 +307,7 @@ suite("test_paimon_catalog",
"p0,external,doris,external_docker,external_docker_
test {
sql """select * from dup_columns_table;"""
- exception "Duplicate column name found: id"
+ exception "Duplicate column name found: ID"
}
sql """ set force_jni_scanner=false; """
@@ -332,4 +332,3 @@ suite("test_paimon_catalog",
"p0,external,doris,external_docker,external_docker_
}
}
-
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
