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 db72b96e524 [fix](be) Harden JNI reader split lifecycle (#65503)
db72b96e524 is described below
commit db72b96e5248666fb56a32beaddaf901fe367366
Author: Gabriel <[email protected]>
AuthorDate: Mon Jul 13 16:29:35 2026 +0800
[fix](be) Harden JNI reader split lifecycle (#65503)
### What problem does this PR solve?
JNI-backed file scans had several lifecycle, compatibility, and
adaptive-batching gaps:
- The scan-level FileScannerV2 selector cannot inspect split-level JNI
metadata. It could therefore route legacy Paimon JNI splits—especially
old-FE splits without `reader_type`—to a scanner that cannot dispatch
them.
- A cancelled query could continue requesting Java batches when every
row in the current batches was filtered out.
- JNI open or close failures could discard cleanup state or mark the
outer scanner closed too early, preventing retained Java resources from
being cleaned up on a retry.
- Adaptive batch-size changes did not consistently reach already-open
JNI scanners and Paimon/Hudi hybrid readers. Paimon's physical reader
cannot be resized after open, so accepting a later logical update could
make the reported batch size diverge from the actual reader.
- End-of-split state could be lost after the JNI scanner closed, making
repeated reads after EOF non-idempotent.
- Paimon options and Hadoop configuration supplied at split level were
not retained as mixed-version fallback values.
This PR hardens those paths by:
- conservatively keeping scan-level JNI compatibility shapes on the
legacy scanner and rejecting unsupported Paimon C++ ranges in
FileScannerV2;
- checking cancellation before fetching another Java batch;
- preserving JNI cleanup state after failures, retaining each Paimon
Java resource until its cleanup succeeds, and allowing scanner-level
close to retry table-reader cleanup;
- seeding the adaptive probe before eager JNI open, forwarding supported
batch-size updates to open JNI and hybrid readers, and preserving
Paimon's initial physical batch size after open;
- preserving EOF state across scanner cleanup; and
- applying explicit precedence between current scan-level Paimon
settings and split-level rolling-upgrade fallbacks.
### Release note
Harden JNI table-reader lifecycle handling, adaptive batching, and
rolling-upgrade compatibility for Paimon and Hudi scans.
### Check List
- [x] Added comments for scanner-selection, lifecycle, adaptive-batch,
and compatibility invariants.
- [x] Added focused unit tests covering cancellation, EOF idempotence,
cleanup failure and retry, adaptive batching, scanner selection, and
Paimon fallback precedence.
- [x] Passed 30 focused ASAN BE unit tests from `FileScannerV2Test.*`,
`JniTableReaderTest.*`, and `PaimonJniReaderTest.*` after addressing the
latest review.
- [x] Passed all 10 `PaimonJniScannerTest` Java tests; checkstyle
reported 0 violations.
- [x] Passed `build-support/check-format.sh` and `git diff --check`.
- [x] Ran changed-file clang-tidy; analysis is blocked by the existing
toolchain missing `stddef.h` and pre-existing repository diagnostics.
---
be/src/exec/operator/file_scan_operator.cpp | 6 +-
be/src/exec/scan/file_scanner_v2.cpp | 20 ++-
be/src/exec/scan/file_scanner_v2.h | 2 +
be/src/format_v2/jni/jni_table_reader.cpp | 149 ++++++++++++----
be/src/format_v2/jni/jni_table_reader.h | 26 ++-
be/src/format_v2/jni/paimon_jni_reader.cpp | 12 +-
be/src/format_v2/jni/paimon_jni_reader.h | 1 +
be/src/format_v2/table/hudi_reader.cpp | 20 ++-
be/src/format_v2/table/hudi_reader.h | 12 ++
be/src/format_v2/table/paimon_reader.cpp | 20 ++-
be/src/format_v2/table/paimon_reader.h | 10 ++
be/src/format_v2/table_reader.h | 6 +-
be/test/exec/scan/file_scanner_v2_test.cpp | 73 ++++++++
be/test/format_v2/jni/jni_table_reader_test.cpp | 191 +++++++++++++++++++++
be/test/format_v2/jni/paimon_jni_reader_test.cpp | 48 ++++++
be/test/format_v2/table/hudi_reader_test.cpp | 9 +
be/test/format_v2/table/paimon_reader_test.cpp | 9 +
.../org/apache/doris/paimon/PaimonJniScanner.java | 13 +-
.../apache/doris/paimon/PaimonJniScannerTest.java | 80 +++++++++
19 files changed, 651 insertions(+), 56 deletions(-)
diff --git a/be/src/exec/operator/file_scan_operator.cpp
b/be/src/exec/operator/file_scan_operator.cpp
index 58aadfc87a6..8121590ebf0 100644
--- a/be/src/exec/operator/file_scan_operator.cpp
+++ b/be/src/exec/operator/file_scan_operator.cpp
@@ -116,10 +116,14 @@ bool
FileScanLocalState::_should_use_file_scanner_v2(const TQueryOptions& query_
const bool is_transactional_hive =
scan_params.__isset.table_format_params &&
scan_params.table_format_params.table_format_type ==
"transactional_hive";
+ // JNI reader selection is stored per split, but this scan-level selector
cannot inspect the
+ // split yet. Older FEs may omit both the scan-level Paimon marker and
split-level reader_type,
+ // so keep JNI scans on V1 until scanner selection can distinguish every
compatibility shape.
return query_options.__isset.enable_file_scanner_v2 &&
query_options.enable_file_scanner_v2 &&
!is_load && scan_params.format_type != TFileFormatType::FORMAT_WAL
&&
scan_params.format_type != TFileFormatType::FORMAT_ES_HTTP &&
- scan_params.format_type != TFileFormatType::FORMAT_LANCE &&
!is_transactional_hive;
+ scan_params.format_type != TFileFormatType::FORMAT_LANCE &&
+ scan_params.format_type != TFileFormatType::FORMAT_JNI &&
!is_transactional_hive;
}
Status FileScanLocalState::_init_scanners(std::list<ScannerSPtr>* scanners) {
diff --git a/be/src/exec/scan/file_scanner_v2.cpp
b/be/src/exec/scan/file_scanner_v2.cpp
index 519237ea85c..6700ed71276 100644
--- a/be/src/exec/scan/file_scanner_v2.cpp
+++ b/be/src/exec/scan/file_scanner_v2.cpp
@@ -212,6 +212,10 @@ Status rewrite_slot_refs_to_global_index(
} // namespace
#ifdef BE_TEST
+FileScannerV2::FileScannerV2(RuntimeState* state, RuntimeProfile* profile,
+ std::unique_ptr<format::TableReader> table_reader)
+ : Scanner(state, profile), _table_reader(std::move(table_reader)) {}
+
Status FileScannerV2::TEST_validate_scan_range(const TFileScanRangeParams&
params,
const TFileRangeDesc& range) {
return _validate_scan_range(params, range);
@@ -407,6 +411,13 @@ Status FileScannerV2::_prepare_next_split(bool* eos) {
DORIS_CHECK(_table_reader != nullptr);
_current_range_path = _current_range.path;
+ const auto format_type = get_range_format_type(*_params,
_current_range);
+ _init_adaptive_batch_size_state(format_type);
+ if (_should_run_adaptive_batch_size()) {
+ // JNI readers open eagerly in prepare_split(). Seed the probe
size first so readers
+ // such as Paimon also use it for their first physical read batch.
+ _table_reader->set_batch_size(_predict_reader_batch_rows());
+ }
std::map<std::string, Field> partition_values;
RETURN_IF_ERROR(_generate_partition_values(_current_range,
&partition_values));
const auto status =
@@ -422,7 +433,6 @@ Status FileScannerV2::_prepare_next_split(bool* eos) {
_state->update_num_finished_scan_range(1);
continue;
}
- _init_adaptive_batch_size_state(get_range_format_type(*_params,
_current_range));
COUNTER_UPDATE(_file_counter, 1);
_has_prepared_split = true;
*eos = false;
@@ -824,7 +834,13 @@ Status FileScannerV2::close(RuntimeState* state) {
return Status::OK();
}
if (_table_reader != nullptr) {
- RETURN_IF_ERROR(_table_reader->close());
+ const auto close_status = _table_reader->close();
+ if (!close_status.ok()) {
+ // Reserve the close attempt with _try_close(), but commit the
scanner-level closed
+ // state only after the retained table reader has completed its
retryable cleanup.
+ _is_closed.store(false);
+ return close_status;
+ }
_report_condition_cache_profile();
_table_reader.reset();
}
diff --git a/be/src/exec/scan/file_scanner_v2.h
b/be/src/exec/scan/file_scanner_v2.h
index 66242b610f4..2bddc5d5e69 100644
--- a/be/src/exec/scan/file_scanner_v2.h
+++ b/be/src/exec/scan/file_scanner_v2.h
@@ -65,6 +65,8 @@ public:
static bool is_supported(const TFileScanRangeParams& params, const
TFileRangeDesc& range);
#ifdef BE_TEST
+ FileScannerV2(RuntimeState* state, RuntimeProfile* profile,
+ std::unique_ptr<format::TableReader> table_reader);
static Status TEST_validate_scan_range(const TFileScanRangeParams& params,
const TFileRangeDesc& range);
static Status TEST_to_file_format(TFileFormatType::type format_type,
diff --git a/be/src/format_v2/jni/jni_table_reader.cpp
b/be/src/format_v2/jni/jni_table_reader.cpp
index 9416659b8f5..3cb105193b5 100644
--- a/be/src/format_v2/jni/jni_table_reader.cpp
+++ b/be/src/format_v2/jni/jni_table_reader.cpp
@@ -36,6 +36,9 @@ Status JniTableReader::init(TableReadOptions&& options) {
}
Status JniTableReader::prepare_split(const SplitReadOptions& options) {
+ // EOF belongs to the previous split. Keep it set after closing that split
so repeated reads
+ // are idempotent, and clear it only when a new split is explicitly
prepared.
+ _eof = false;
_current_range = options.current_range;
RETURN_IF_ERROR(validate_scan_range(options.current_range));
RETURN_IF_ERROR(TableReader::prepare_split(options));
@@ -68,13 +71,21 @@ Status JniTableReader::get_block(Block* output_block, bool*
eos) {
return _read_table_level_count(output_block, eos);
}
- DORIS_CHECK(_scanner_opened);
if (_eof) {
*eos = true;
return Status::OK();
}
+ DORIS_CHECK(_scanner_opened);
while (true) {
+ // JNI readers can loop internally when conjuncts filter every Java
batch. Mirror the base
+ // TableReader cancellation contract so a cancelled query does not
drain the whole split.
+ if (_io_ctx != nullptr && _io_ctx->should_stop) {
+ _eof = true;
+ RETURN_IF_ERROR(_close_jni_scanner());
+ *eos = true;
+ return Status::OK();
+ }
size_t current_rows = 0;
bool current_eof = false;
// get next block data from Java scanner, and fill the data to
_jni_block_template
@@ -282,41 +293,43 @@ int64_t JniTableReader::self_split_weight() const {
return _current_range.__isset.self_split_weight ?
_current_range.self_split_weight : -1;
}
-Status JniTableReader::close() {
- if (_closed) {
- return Status::OK();
+bool JniTableReader::_reserve_split_profile_publication() {
+ if (_split_profile_published) {
+ return false;
}
- _closed = true;
- RETURN_IF_ERROR(_close_jni_scanner());
- return TableReader::close();
+ _split_profile_published = true;
+ return true;
}
-Status JniTableReader::_close_jni_scanner() {
- if (!_scanner_opened) {
- JNIEnv* env = nullptr;
- if (!_jni_scanner_obj.uninitialized()) {
- RETURN_IF_ERROR(Jni::Env::Get(&env));
- }
- _reset_split_state(env);
- return Status::OK();
+void JniTableReader::_publish_split_profile(JNIEnv* env) {
+ // Cleanup can fail while the Java scanner and split watchers must remain
available for a
+ // retry. Reserve profile publication separately so a retry only repeats
resource cleanup.
+ if (!_reserve_split_profile_publication()) {
+ return;
}
- JNIEnv* env = nullptr;
- RETURN_IF_ERROR(Jni::Env::Get(&env));
if (_scanner_profile != nullptr) {
COUNTER_UPDATE(_open_scanner_time, _jni_scanner_open_watcher);
COUNTER_UPDATE(_fill_block_time, _fill_block_watcher);
}
- RETURN_ERROR_IF_EXC(env);
jlong append_data_time = 0;
- RETURN_IF_ERROR(_jni_scanner_obj.call_long_method(env,
_jni_scanner_get_append_data_time)
- .call(&append_data_time));
+ const auto append_time_status =
+ _jni_scanner_obj.call_long_method(env,
_jni_scanner_get_append_data_time)
+ .call(&append_data_time);
jlong create_vector_table_time = 0;
- RETURN_IF_ERROR(
+ const auto create_table_time_status =
_jni_scanner_obj.call_long_method(env,
_jni_scanner_get_create_vector_table_time)
- .call(&create_vector_table_time));
- if (_scanner_profile != nullptr) {
+ .call(&create_vector_table_time);
+ if (!append_time_status.ok()) {
+ LOG(WARNING) << "failed to collect JNI append-data time during close: "
+ << append_time_status;
+ }
+ if (!create_table_time_status.ok()) {
+ LOG(WARNING) << "failed to collect JNI vector-table time during close:
"
+ << create_table_time_status;
+ }
+ if (_scanner_profile != nullptr && append_time_status.ok() &&
create_table_time_status.ok()) {
COUNTER_UPDATE(_java_append_data_time, append_data_time);
COUNTER_UPDATE(_java_create_vector_table_time,
create_vector_table_time);
COUNTER_UPDATE(_java_scan_time,
@@ -326,13 +339,50 @@ Status JniTableReader::_close_jni_scanner() {
self_split_weight());
}
_collect_jni_scanner_profile(env);
+}
+
+Status JniTableReader::close() {
+ if (_closed) {
+ return Status::OK();
+ }
+ auto close_status = _close_jni_scanner();
+ auto table_status = TableReader::close();
+ if (close_status.ok() && !table_status.ok()) {
+ close_status = std::move(table_status);
+ }
+ if (close_status.ok()) {
+ _closed = true;
+ }
+ return close_status;
+}
+
+Status JniTableReader::_close_jni_scanner() {
+ if (!_scanner_opened) {
+ JNIEnv* env = nullptr;
+ if (!_jni_scanner_obj.uninitialized()) {
+ RETURN_IF_ERROR(Jni::Env::Get(&env));
+ }
+ _reset_split_state(env);
+ return Status::OK();
+ }
+
+ JNIEnv* env = nullptr;
+ RETURN_IF_ERROR(Jni::Env::Get(&env));
+ _publish_split_profile(env);
// _fill_jni_block may fail before releasing the current Java table.
JniScanner::releaseTable()
- // is idempotent, so closing the split always releases it.
- RETURN_IF_ERROR(_jni_scanner_obj.call_void_method(env,
_jni_scanner_release_table).call());
- RETURN_IF_ERROR(_jni_scanner_obj.call_void_method(env,
_jni_scanner_close).call());
- _reset_split_state(env);
- return Status::OK();
+ // is idempotent, so closing the split always releases it. Java close must
still run if that
+ // release fails; otherwise connector resources such as JDBC connections
can leak.
+ auto cleanup_status = _jni_scanner_obj.call_void_method(env,
_jni_scanner_release_table).call();
+ auto java_close_status = _jni_scanner_obj.call_void_method(env,
_jni_scanner_close).call();
+ if (cleanup_status.ok() && !java_close_status.ok()) {
+ cleanup_status = std::move(java_close_status);
+ }
+ if (cleanup_status.ok()) {
+ // Keep the Java object and opened state on failure so close() can
retry the cleanup.
+ _reset_split_state(env);
+ }
+ return cleanup_status;
}
void JniTableReader::_reset_split_state(JNIEnv* env) {
@@ -341,13 +391,13 @@ void JniTableReader::_reset_split_state(JNIEnv* env) {
_jni_scanner_obj.reset(env);
}
_scanner_opened = false;
- _eof = false;
_scanner_params.clear();
_jni_columns.clear();
_jni_block_template.clear();
_jni_scanner_open_watcher = 0;
_java_scan_watcher = 0;
_fill_block_watcher = 0;
+ _split_profile_published = false;
}
Status JniTableReader::_open_jni_scanner() {
@@ -370,14 +420,47 @@ Status JniTableReader::_open_jni_scanner() {
SCOPED_RAW_TIMER(&_jni_scanner_open_watcher);
RETURN_IF_ERROR(_register_jni_class_functions_once(env));
RETURN_IF_ERROR(_create_jni_scanner_object(env,
cast_set<int>(_batch_size)));
- // call open() method in JAVA side.
- RETURN_IF_ERROR(_jni_scanner_obj.call_void_method(env,
_jni_scanner_open).call());
- RETURN_ERROR_IF_EXC(env);
-
+ // Once the Java object exists, close it even if open() fails partway
through initialization.
+ // Connector implementations may already own streams, off-heap tables, or
JDBC connections.
_scanner_opened = true;
+ // call open() method in JAVA side.
+ const auto open_status = _jni_scanner_obj.call_void_method(env,
_jni_scanner_open).call();
+ if (!open_status.ok()) {
+ const auto close_status = _close_jni_scanner();
+ if (!close_status.ok()) {
+ LOG(WARNING) << "failed to clean up JNI scanner after open
failure: " << close_status;
+ }
+ return open_status;
+ }
return Status::OK();
}
+void JniTableReader::set_batch_size(size_t batch_size) {
+ if (_scanner_opened && !supports_batch_size_update_after_open()) {
+ // Some connectors bake the constructor batch size into an
already-open physical reader.
+ // Keep C++ and Java on that initial size instead of pretending a
later resize took effect.
+ return;
+ }
+ TableReader::set_batch_size(batch_size);
+ if (!_scanner_opened) {
+ return;
+ }
+ const auto status = _set_open_scanner_batch_size(_batch_size);
+ if (!status.ok()) {
+ // Adaptive batch sizing is an optimization. Keep the scanner usable
with its previous
+ // size if Java rejects a mid-split update, but surface the failure
for diagnosis.
+ LOG(WARNING) << "failed to update JNI scanner batch size: " << status;
+ }
+}
+
+Status JniTableReader::_set_open_scanner_batch_size(size_t batch_size) {
+ JNIEnv* env = nullptr;
+ RETURN_IF_ERROR(Jni::Env::Get(&env));
+ return _jni_scanner_obj.call_void_method(env, _jni_scanner_set_batch_size)
+ .with_arg(cast_set<int>(batch_size))
+ .call();
+}
+
void JniTableReader::_prepare_jni_scanner_schema() {
std::vector<std::string> required_fields;
std::vector<std::string> column_types;
diff --git a/be/src/format_v2/jni/jni_table_reader.h
b/be/src/format_v2/jni/jni_table_reader.h
index f9c854e1971..24f84ca8d62 100644
--- a/be/src/format_v2/jni/jni_table_reader.h
+++ b/be/src/format_v2/jni/jni_table_reader.h
@@ -51,6 +51,20 @@ public:
Status get_block(Block* block, bool* eos) override;
Status abort_split() override;
Status close() override;
+ void set_batch_size(size_t batch_size) override;
+
+#ifdef BE_TEST
+ void TEST_set_split_state(bool scanner_opened, bool eof) {
+ _scanner_opened = scanner_opened;
+ _eof = eof;
+ if (!scanner_opened) {
+ _split_profile_published = false;
+ }
+ }
+ bool TEST_scanner_opened() const { return _scanner_opened; }
+ bool TEST_eof() const { return _eof; }
+ bool TEST_closed() const { return _closed; }
+#endif
protected:
// Subclasses should implement these methods to specify the Java scanner
class
@@ -63,6 +77,12 @@ protected:
virtual Status finalize_jni_block(Block* jni_block, Block* output_block,
size_t* rows);
// used for profile
virtual int64_t self_split_weight() const;
+ virtual Status _get_next_jni_block(size_t* rows, bool* eof);
+ virtual Status _close_jni_scanner();
+ virtual Status _set_open_scanner_batch_size(size_t batch_size);
+ virtual bool supports_batch_size_update_after_open() const { return true; }
+ virtual Status _open_jni_scanner();
+ bool _reserve_split_profile_publication();
const std::vector<JniColumn>& jni_columns() const { return _jni_columns; }
TFileRangeDesc _current_range;
@@ -71,18 +91,15 @@ private:
void _init_profile();
std::string _connector_name() const;
// open
- Status _open_jni_scanner();
void _reset_split_state(JNIEnv* env);
void _prepare_jni_scanner_schema();
Status _register_jni_class_functions_once(JNIEnv* env);
Status _create_jni_scanner_object(JNIEnv* env, int batch_size);
// get_next
- Status _get_next_jni_block(size_t* rows, bool* eof);
Status _fill_jni_block(JniDataBridge::TableMetaAddress& table_meta, size_t
num_rows);
Status _get_statistics(JNIEnv* env, std::map<std::string, std::string>*
result);
void _collect_jni_scanner_profile(JNIEnv* env);
-
- Status _close_jni_scanner();
+ void _publish_split_profile(JNIEnv* env);
std::map<std::string, std::string> _scanner_params;
std::vector<JniColumn> _jni_columns;
@@ -91,6 +108,7 @@ private:
bool _closed = false;
bool _scanner_opened = false;
bool _eof = false;
+ bool _split_profile_published = false;
RuntimeProfile::Counter* _open_scanner_time = nullptr;
RuntimeProfile::Counter* _java_scan_time = nullptr;
diff --git a/be/src/format_v2/jni/paimon_jni_reader.cpp
b/be/src/format_v2/jni/paimon_jni_reader.cpp
index ca1e602cdfe..47c0ef6c7bc 100644
--- a/be/src/format_v2/jni/paimon_jni_reader.cpp
+++ b/be/src/format_v2/jni/paimon_jni_reader.cpp
@@ -99,6 +99,11 @@ Status
PaimonJniReader::build_scanner_params(std::map<std::string, std::string>*
for (const auto& kv : _scan_params->paimon_options) {
(*params)[std::string(PAIMON_OPTION_PREFIX) + kv.first] =
kv.second;
}
+ } else if (paimon_params.__isset.paimon_options) {
+ // Rolling upgrades can pair this BE with an older FE that only sends
options per split.
+ for (const auto& kv : paimon_params.paimon_options) {
+ (*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);
@@ -118,10 +123,13 @@ Status
PaimonJniReader::build_scanner_params(std::map<std::string, std::string>*
for (const auto& kv : _scan_params->properties) {
(*params)[std::string(HADOOP_OPTION_PREFIX) + kv.first] =
kv.second;
}
+ } else if (paimon_params.__isset.hadoop_conf) {
+ for (const auto& kv : paimon_params.hadoop_conf) {
+ (*params)[std::string(HADOOP_OPTION_PREFIX) + kv.first] =
kv.second;
+ }
}
// TODO: Remove legacy split-level paimon_predicate, paimon_options and
hadoop_conf from thrift
- // after all readers stop using them. Predicate keeps the split-level
fallback for rolling
- // upgrade compatibility with old FE paths that did not send scan-level
paimon_predicate.
+ // after the minimum supported FE always sends their scan-level
replacements.
return Status::OK();
}
diff --git a/be/src/format_v2/jni/paimon_jni_reader.h
b/be/src/format_v2/jni/paimon_jni_reader.h
index 30788081565..26767458b42 100644
--- a/be/src/format_v2/jni/paimon_jni_reader.h
+++ b/be/src/format_v2/jni/paimon_jni_reader.h
@@ -50,6 +50,7 @@ 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;
+ bool supports_batch_size_update_after_open() const override { return
false; }
private:
static std::string build_default_io_manager_tmp_dirs(const
std::vector<StorePath>& store_paths);
diff --git a/be/src/format_v2/table/hudi_reader.cpp
b/be/src/format_v2/table/hudi_reader.cpp
index 3600b0ff706..f37d24bd888 100644
--- a/be/src/format_v2/table/hudi_reader.cpp
+++ b/be/src/format_v2/table/hudi_reader.cpp
@@ -91,6 +91,16 @@ Status HudiHybridReader::close() {
return close_status;
}
+void HudiHybridReader::set_batch_size(size_t batch_size) {
+ format::TableReader::set_batch_size(batch_size);
+ if (_native_reader != nullptr) {
+ _native_reader->set_batch_size(_batch_size);
+ }
+ if (_jni_reader != nullptr) {
+ _jni_reader->set_batch_size(_batch_size);
+ }
+}
+
Status HudiHybridReader::_ensure_current_split_reader(const
format::SplitReadOptions& options) {
DORIS_CHECK(_scan_params != nullptr);
if (_is_jni_split(*_scan_params, options.current_range)) {
@@ -116,7 +126,7 @@ Status
HudiHybridReader::_init_child_reader(format::TableReader* reader,
DORIS_CHECK(reader != nullptr);
VExprContextSPtrs conjuncts;
RETURN_IF_ERROR(_clone_conjuncts(&conjuncts));
- return reader->init({
+ RETURN_IF_ERROR(reader->init({
.projected_columns = _projected_columns,
.conjuncts = std::move(conjuncts),
.format = file_format,
@@ -126,7 +136,13 @@ Status
HudiHybridReader::_init_child_reader(format::TableReader* reader,
.scanner_profile = _scanner_profile,
.push_down_agg_type = _push_down_agg_type,
.condition_cache_digest = _condition_cache_digest,
- });
+ }));
+ // Zero means no adaptive prediction has been produced yet. Preserve the
child's normal
+ // runtime default until FileScannerV2 supplies the first positive
prediction.
+ if (_batch_size > 0) {
+ reader->set_batch_size(_batch_size);
+ }
+ return Status::OK();
}
Status HudiHybridReader::_clone_conjuncts(VExprContextSPtrs* conjuncts) const {
diff --git a/be/src/format_v2/table/hudi_reader.h
b/be/src/format_v2/table/hudi_reader.h
index 58f222c67ab..e22c6bd866f 100644
--- a/be/src/format_v2/table/hudi_reader.h
+++ b/be/src/format_v2/table/hudi_reader.h
@@ -18,6 +18,7 @@
#pragma once
#include <memory>
+#include <utility>
#include <vector>
#include "format_v2/table_reader.h"
@@ -61,6 +62,17 @@ public:
bool current_split_pruned() const override;
Status abort_split() override;
Status close() override;
+ void set_batch_size(size_t batch_size) override;
+
+#ifdef BE_TEST
+ void TEST_install_batch_size_children() {
+ _native_reader = std::make_unique<format::TableReader>();
+ _jni_reader = std::make_unique<format::TableReader>();
+ }
+ std::pair<size_t, size_t> TEST_child_batch_sizes() const {
+ return {_native_reader->TEST_batch_size(),
_jni_reader->TEST_batch_size()};
+ }
+#endif
private:
Status _ensure_current_split_reader(const format::SplitReadOptions&
options);
diff --git a/be/src/format_v2/table/paimon_reader.cpp
b/be/src/format_v2/table/paimon_reader.cpp
index c2114968946..236daac118a 100644
--- a/be/src/format_v2/table/paimon_reader.cpp
+++ b/be/src/format_v2/table/paimon_reader.cpp
@@ -116,6 +116,16 @@ Status PaimonHybridReader::close() {
return close_status;
}
+void PaimonHybridReader::set_batch_size(size_t batch_size) {
+ format::TableReader::set_batch_size(batch_size);
+ if (_native_reader != nullptr) {
+ _native_reader->set_batch_size(_batch_size);
+ }
+ if (_jni_reader != nullptr) {
+ _jni_reader->set_batch_size(_batch_size);
+ }
+}
+
Status PaimonHybridReader::_ensure_current_split_reader(const
format::SplitReadOptions& options) {
if (_is_jni_split(options.current_range)) {
DCHECK(options.current_split_format == format::FileFormat::JNI);
@@ -144,7 +154,7 @@ Status
PaimonHybridReader::_init_child_reader(format::TableReader* reader,
DORIS_CHECK(reader != nullptr);
VExprContextSPtrs conjuncts;
RETURN_IF_ERROR(_clone_conjuncts(&conjuncts));
- return reader->init({
+ RETURN_IF_ERROR(reader->init({
.projected_columns = _projected_columns,
.conjuncts = std::move(conjuncts),
.format = file_format,
@@ -154,7 +164,13 @@ Status
PaimonHybridReader::_init_child_reader(format::TableReader* reader,
.scanner_profile = _scanner_profile,
.push_down_agg_type = _push_down_agg_type,
.condition_cache_digest = _condition_cache_digest,
- });
+ }));
+ // Zero means no adaptive prediction has been produced yet. Preserve the
child's normal
+ // runtime default until FileScannerV2 supplies the first positive
prediction.
+ if (_batch_size > 0) {
+ reader->set_batch_size(_batch_size);
+ }
+ return Status::OK();
}
Status PaimonHybridReader::_clone_conjuncts(VExprContextSPtrs* conjuncts)
const {
diff --git a/be/src/format_v2/table/paimon_reader.h
b/be/src/format_v2/table/paimon_reader.h
index ae179e6135a..fc94777b6da 100644
--- a/be/src/format_v2/table/paimon_reader.h
+++ b/be/src/format_v2/table/paimon_reader.h
@@ -17,6 +17,8 @@
#pragma once
+#include <utility>
+
#include "format_v2/table_reader.h"
namespace doris {
@@ -66,6 +68,7 @@ public:
bool current_split_pruned() const override;
Status abort_split() override;
Status close() override;
+ void set_batch_size(size_t batch_size) override;
#ifdef BE_TEST
static bool TEST_is_jni_split(const TFileRangeDesc& range) { return
_is_jni_split(range); }
@@ -73,6 +76,13 @@ public:
format::FileFormat* file_format) {
return _to_file_format(range, file_format);
}
+ void TEST_install_batch_size_children() {
+ _native_reader = std::make_unique<format::TableReader>();
+ _jni_reader = std::make_unique<format::TableReader>();
+ }
+ std::pair<size_t, size_t> TEST_child_batch_sizes() const {
+ return {_native_reader->TEST_batch_size(),
_jni_reader->TEST_batch_size()};
+ }
#endif
private:
diff --git a/be/src/format_v2/table_reader.h b/be/src/format_v2/table_reader.h
index 643fafc1fe4..b8457ee9b4a 100644
--- a/be/src/format_v2/table_reader.h
+++ b/be/src/format_v2/table_reader.h
@@ -176,13 +176,17 @@ public:
// FileScannerV2 adjusts this before each get_block() using an adaptive
bytes-per-row estimate.
// Store it here as well as forwarding to the current reader so newly
opened split readers start
// with the latest predicted batch size.
- void set_batch_size(size_t batch_size) {
+ virtual void set_batch_size(size_t batch_size) {
_batch_size = std::max<size_t>(1, batch_size);
if (_data_reader.reader != nullptr) {
_data_reader.reader->set_batch_size(_batch_size);
}
}
+#ifdef BE_TEST
+ size_t TEST_batch_size() const { return _batch_size; }
+#endif
+
// Prepare for reading a new split/task.
// 1. Pass a new split/task to reader, which will be used in subsequent
open_reader() to initialize the underlying file reader.
// 2. Parse delete predicates from split/task information, which will be
used for later dynamic filtering and delete handling.
diff --git a/be/test/exec/scan/file_scanner_v2_test.cpp
b/be/test/exec/scan/file_scanner_v2_test.cpp
index 2ab86072327..abd3d1e76d7 100644
--- a/be/test/exec/scan/file_scanner_v2_test.cpp
+++ b/be/test/exec/scan/file_scanner_v2_test.cpp
@@ -63,6 +63,44 @@ TFileRangeDesc hudi_range_with_delta_logs() {
return range;
}
+TFileRangeDesc paimon_cpp_jni_range() {
+ auto range = range_with_format("paimon", TFileFormatType::FORMAT_JNI);
+ TPaimonFileDesc paimon_params;
+ paimon_params.__set_reader_type(TPaimonReaderType::PAIMON_CPP);
+ range.table_format_params.__set_paimon_params(std::move(paimon_params));
+ return range;
+}
+
+TFileRangeDesc legacy_paimon_jni_range_without_reader_type() {
+ auto range = range_with_format("paimon", TFileFormatType::FORMAT_JNI);
+ TPaimonFileDesc paimon_params;
+ paimon_params.__set_paimon_split("legacy-split");
+ paimon_params.__set_paimon_predicate("legacy-predicate");
+ range.table_format_params.__set_paimon_params(std::move(paimon_params));
+ return range;
+}
+
+struct RetryableCloseState {
+ int close_calls = 0;
+};
+
+class RetryableCloseTableReader final : public format::TableReader {
+public:
+ explicit RetryableCloseTableReader(std::shared_ptr<RetryableCloseState>
state)
+ : _state(std::move(state)) {}
+
+ Status close() override {
+ ++_state->close_calls;
+ if (_state->close_calls == 1) {
+ return Status::InternalError("injected table reader close
failure");
+ }
+ return Status::OK();
+ }
+
+private:
+ std::shared_ptr<RetryableCloseState> _state;
+};
+
VExprSPtr slot_ref(int slot_id, int column_id, DataTypePtr type, const
std::string& name) {
return VSlotRef::create_shared(slot_id, column_id, -1, std::move(type),
name);
}
@@ -216,6 +254,41 @@ TEST(FileScannerV2Test,
FileScanLocalStateSelectsV2ForSupportedQueriesOnly) {
EXPECT_FALSE(FileScanLocalState::TEST_should_use_file_scanner_v2(query_options,
false, params));
}
+TEST(FileScannerV2Test, JniCompatibilityShapesForceLegacyScanner) {
+ TQueryOptions query_options;
+ query_options.__set_enable_file_scanner_v2(true);
+ query_options.__set_enable_paimon_cpp_reader(true);
+
+ TFileScanRangeParams params;
+ params.__set_format_type(TFileFormatType::FORMAT_JNI);
+ // Rolling upgrades may carry the only Paimon marker and reader type on
each split. Since the
+ // scan-level selector cannot inspect that split yet, JNI scans
conservatively stay on V1.
+
EXPECT_FALSE(FileScanLocalState::TEST_should_use_file_scanner_v2(query_options,
false, params));
+ EXPECT_FALSE(FileScannerV2::is_supported(params, paimon_cpp_jni_range()));
+
+ // Older FEs can omit reader_type. The legacy scanner interprets this as
Paimon JNI when the C++
+ // reader is disabled, so the scan-level choice must still stay on V1.
+ query_options.__set_enable_paimon_cpp_reader(false);
+
EXPECT_FALSE(FileScanLocalState::TEST_should_use_file_scanner_v2(query_options,
false, params));
+ EXPECT_FALSE(
+ FileScannerV2::is_supported(params,
legacy_paimon_jni_range_without_reader_type()));
+}
+
+TEST(FileScannerV2Test, FailedTableReaderCloseCanBeRetriedThroughScanner) {
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ RuntimeProfile profile("file_scanner_v2_close_retry");
+ auto close_state = std::make_shared<RetryableCloseState>();
+ FileScannerV2 scanner(&state, &profile,
+
std::make_unique<RetryableCloseTableReader>(close_state));
+
+ EXPECT_FALSE(scanner.close(&state).ok());
+ EXPECT_EQ(close_state->close_calls, 1);
+ EXPECT_TRUE(scanner.close(&state).ok());
+ EXPECT_EQ(close_state->close_calls, 2);
+ EXPECT_TRUE(scanner.close(&state).ok());
+ EXPECT_EQ(close_state->close_calls, 2);
+}
+
// Scenario: Once FileScannerV2 is selected, an unsupported range must fail
instead of falling back
// to FileScanner.
TEST(FileScannerV2Test, ValidateScanRangeRejectsUnsupportedRange) {
diff --git a/be/test/format_v2/jni/jni_table_reader_test.cpp
b/be/test/format_v2/jni/jni_table_reader_test.cpp
new file mode 100644
index 00000000000..1c5788b1df7
--- /dev/null
+++ b/be/test/format_v2/jni/jni_table_reader_test.cpp
@@ -0,0 +1,191 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "format_v2/jni/jni_table_reader.h"
+
+#include <gtest/gtest.h>
+
+#include <map>
+#include <memory>
+#include <string>
+#include <vector>
+
+#include "io/io_common.h"
+
+namespace doris::format {
+namespace {
+
+class FakeJniTableReader final : public JniTableReader {
+public:
+ int get_next_calls = 0;
+ int close_calls = 0;
+ int64_t close_profile_delta = 0;
+ RuntimeProfile::Counter* close_profile_counter = nullptr;
+ std::vector<size_t> open_batch_sizes;
+ std::vector<size_t> propagated_batch_sizes;
+ std::vector<Status> close_results;
+ bool next_eof = false;
+
+protected:
+ std::string connector_class() const override { return
"test/FakeJniScanner"; }
+
+ Status build_scanner_params(std::map<std::string, std::string>* params)
const override {
+ params->clear();
+ return Status::OK();
+ }
+
+ Status _get_next_jni_block(size_t* rows, bool* eof) override {
+ ++get_next_calls;
+ *rows = 0;
+ *eof = next_eof;
+ return Status::OK();
+ }
+
+ Status _close_jni_scanner() override {
+ if (!TEST_scanner_opened()) {
+ return Status::OK();
+ }
+ ++close_calls;
+ if (_reserve_split_profile_publication() && close_profile_counter !=
nullptr) {
+ COUNTER_UPDATE(close_profile_counter, close_profile_delta);
+ }
+ Status status = Status::OK();
+ if (static_cast<size_t>(close_calls) <= close_results.size()) {
+ status = close_results[close_calls - 1];
+ }
+ if (status.ok()) {
+ TEST_set_split_state(false, TEST_eof());
+ }
+ return status;
+ }
+
+ Status _set_open_scanner_batch_size(size_t batch_size) override {
+ propagated_batch_sizes.push_back(batch_size);
+ return Status::OK();
+ }
+
+ Status _open_jni_scanner() override {
+ open_batch_sizes.push_back(TEST_batch_size());
+ TEST_set_split_state(true, false);
+ return Status::OK();
+ }
+};
+
+Status init_reader(FakeJniTableReader* reader, const
std::shared_ptr<io::IOContext>& io_ctx,
+ RuntimeProfile* scanner_profile = nullptr) {
+ return reader->init({
+ .projected_columns = {},
+ .conjuncts = {},
+ .format = FileFormat::JNI,
+ .scan_params = nullptr,
+ .io_ctx = io_ctx,
+ .runtime_state = nullptr,
+ .scanner_profile = scanner_profile,
+ });
+}
+
+TEST(JniTableReaderTest, CancellationStopsBeforeFetchingAnotherJavaBatch) {
+ auto io_ctx = std::make_shared<io::IOContext>();
+ FakeJniTableReader reader;
+ ASSERT_TRUE(init_reader(&reader, io_ctx).ok());
+ reader.TEST_set_split_state(true, false);
+ io_ctx->should_stop = true;
+
+ Block block;
+ bool eos = false;
+ ASSERT_TRUE(reader.get_block(&block, &eos).ok());
+ EXPECT_TRUE(eos);
+ EXPECT_EQ(reader.get_next_calls, 0);
+ EXPECT_EQ(reader.close_calls, 1);
+}
+
+TEST(JniTableReaderTest, EndOfSplitRemainsIdempotentAfterScannerClose) {
+ FakeJniTableReader reader;
+ ASSERT_TRUE(init_reader(&reader, nullptr).ok());
+ reader.TEST_set_split_state(true, false);
+ reader.next_eof = true;
+
+ Block block;
+ bool eos = false;
+ ASSERT_TRUE(reader.get_block(&block, &eos).ok());
+ EXPECT_TRUE(eos);
+ EXPECT_TRUE(reader.TEST_eof());
+ EXPECT_FALSE(reader.TEST_scanner_opened());
+
+ eos = false;
+ ASSERT_TRUE(reader.get_block(&block, &eos).ok());
+ EXPECT_TRUE(eos);
+ EXPECT_EQ(reader.get_next_calls, 1);
+ EXPECT_EQ(reader.close_calls, 1);
+}
+
+TEST(JniTableReaderTest, FailedCloseCanBeRetried) {
+ RuntimeProfile profile("FailedCloseCanBeRetried");
+ FakeJniTableReader reader;
+ reader.close_profile_counter =
profile.add_counter("PublishedCloseProfile", TUnit::UNIT);
+ reader.close_profile_delta = 17;
+ ASSERT_TRUE(init_reader(&reader, nullptr, &profile).ok());
+ reader.TEST_set_split_state(true, false);
+ reader.close_results = {Status::InternalError("injected close failure"),
Status::OK()};
+
+ EXPECT_FALSE(reader.close().ok());
+ EXPECT_FALSE(reader.TEST_closed());
+ EXPECT_TRUE(reader.TEST_scanner_opened());
+ EXPECT_EQ(reader.close_calls, 1);
+ EXPECT_EQ(reader.close_profile_counter->value(), 17);
+
+ EXPECT_TRUE(reader.close().ok());
+ EXPECT_TRUE(reader.TEST_closed());
+ EXPECT_EQ(reader.close_calls, 2);
+ EXPECT_EQ(reader.close_profile_counter->value(), 17);
+}
+
+TEST(JniTableReaderTest, AdaptiveBatchSizeUpdatesAnOpenJavaScanner) {
+ FakeJniTableReader reader;
+ ASSERT_TRUE(init_reader(&reader, nullptr).ok());
+
+ reader.set_batch_size(17);
+ EXPECT_EQ(reader.TEST_batch_size(), 17);
+ EXPECT_TRUE(reader.propagated_batch_sizes.empty());
+
+ reader.TEST_set_split_state(true, false);
+ reader.set_batch_size(33);
+ EXPECT_EQ(reader.TEST_batch_size(), 33);
+ EXPECT_EQ(reader.propagated_batch_sizes, std::vector<size_t>({33}));
+}
+
+TEST(JniTableReaderTest, AdaptiveProbeSetBeforePrepareControlsFirstJniOpen) {
+ FakeJniTableReader reader;
+ ASSERT_TRUE(init_reader(&reader, nullptr).ok());
+
+ reader.set_batch_size(32);
+ ASSERT_TRUE(reader.prepare_split({
+ .partition_values = {},
+ .partition_prune_conjuncts = {},
+ .cache = nullptr,
+ .current_range = {},
+ .current_split_format =
FileFormat::JNI,
+ .global_rowid_context =
std::nullopt,
+ })
+ .ok());
+
+ EXPECT_EQ(reader.open_batch_sizes, std::vector<size_t>({32}));
+ EXPECT_TRUE(reader.TEST_scanner_opened());
+}
+
+} // namespace
+} // namespace doris::format
diff --git a/be/test/format_v2/jni/paimon_jni_reader_test.cpp
b/be/test/format_v2/jni/paimon_jni_reader_test.cpp
index 08fd333abd3..570977a5f7e 100644
--- a/be/test/format_v2/jni/paimon_jni_reader_test.cpp
+++ b/be/test/format_v2/jni/paimon_jni_reader_test.cpp
@@ -123,5 +123,53 @@ TEST(PaimonJniReaderTest,
RejectsMissingPredicateFromBothProtocolLocations) {
EXPECT_NE(status.to_string().find("missing paimon_predicate"),
std::string::npos);
}
+TEST(PaimonJniReaderTest, FallsBackToLegacySplitOptionsAndHadoopConf) {
+ auto range = make_paimon_jni_range();
+ auto& paimon_params = range.table_format_params.paimon_params;
+ paimon_params.__set_paimon_predicate("legacy-predicate");
+ paimon_params.__set_paimon_options({{"legacy-option", "legacy-value"}});
+ paimon_params.__set_hadoop_conf({{"fs.defaultFS", "hdfs://legacy"}});
+
+ auto scan_params = make_scan_params();
+ PaimonJniReader reader;
+ ASSERT_TRUE(init_reader(&reader, &scan_params).ok());
+
+ std::map<std::string, std::string> params;
+ ASSERT_TRUE(build_params(&reader, range, ¶ms).ok());
+ EXPECT_EQ(params["paimon.legacy-option"], "legacy-value");
+ EXPECT_EQ(params["hadoop.fs.defaultFS"], "hdfs://legacy");
+}
+
+TEST(PaimonJniReaderTest, ScanLevelOptionsOverrideLegacySplitFallbacks) {
+ auto range = make_paimon_jni_range();
+ auto& paimon_params = range.table_format_params.paimon_params;
+ paimon_params.__set_paimon_predicate("legacy-predicate");
+ paimon_params.__set_paimon_options({{"source", "legacy"}});
+ paimon_params.__set_hadoop_conf({{"source", "legacy"}});
+
+ auto scan_params = make_scan_params();
+ scan_params.__set_paimon_options({{"source", "scan"}});
+ scan_params.__set_properties({{"source", "scan"}});
+ PaimonJniReader reader;
+ ASSERT_TRUE(init_reader(&reader, &scan_params).ok());
+
+ std::map<std::string, std::string> params;
+ ASSERT_TRUE(build_params(&reader, range, ¶ms).ok());
+ EXPECT_EQ(params["paimon.source"], "scan");
+ EXPECT_EQ(params["hadoop.source"], "scan");
+}
+
+TEST(PaimonJniReaderTest, KeepsInitialPhysicalBatchSizeAfterOpen) {
+ PaimonJniReader reader;
+ reader.set_batch_size(32);
+ EXPECT_EQ(reader.TEST_batch_size(), 32);
+
+ // Paimon copies the constructor size into the RecordReader during Java
open. A later predictor
+ // result cannot resize that physical reader, so keep the initial probe
size for the split.
+ reader.TEST_set_split_state(true, false);
+ reader.set_batch_size(1);
+ EXPECT_EQ(reader.TEST_batch_size(), 32);
+}
+
} // namespace
} // namespace doris::format::paimon
diff --git a/be/test/format_v2/table/hudi_reader_test.cpp
b/be/test/format_v2/table/hudi_reader_test.cpp
index 125183cd7a6..77cd75bc948 100644
--- a/be/test/format_v2/table/hudi_reader_test.cpp
+++ b/be/test/format_v2/table/hudi_reader_test.cpp
@@ -178,5 +178,14 @@ TEST(HudiReaderTest,
ResetsSplitSchemaIdBeforePreparingNextSplit) {
EXPECT_EQ(reader.TEST_mapping_mode(), TableColumnMappingMode::BY_NAME);
}
+TEST(HudiHybridReaderTest, AdaptiveBatchSizeReachesBothChildReaders) {
+ hudi::HudiHybridReader reader;
+ reader.TEST_install_batch_size_children();
+ reader.set_batch_size(123);
+ const auto child_batch_sizes = reader.TEST_child_batch_sizes();
+ EXPECT_EQ(child_batch_sizes.first, 123);
+ EXPECT_EQ(child_batch_sizes.second, 123);
+}
+
} // namespace
} // namespace doris::format
diff --git a/be/test/format_v2/table/paimon_reader_test.cpp
b/be/test/format_v2/table/paimon_reader_test.cpp
index f1c9196e733..a56ebcc5e87 100644
--- a/be/test/format_v2/table/paimon_reader_test.cpp
+++ b/be/test/format_v2/table/paimon_reader_test.cpp
@@ -637,6 +637,15 @@ TEST(PaimonHybridReaderTest,
ConvertsNativeSplitFileFormat) {
EXPECT_NE(std::string::npos, status.to_string().find("Unsupported native
Paimon file format"));
}
+TEST(PaimonHybridReaderTest, AdaptiveBatchSizeReachesBothChildReaders) {
+ paimon::PaimonHybridReader reader;
+ reader.TEST_install_batch_size_children();
+ reader.set_batch_size(321);
+ const auto child_batch_sizes = reader.TEST_child_batch_sizes();
+ EXPECT_EQ(child_batch_sizes.first, 321);
+ EXPECT_EQ(child_batch_sizes.second, 321);
+}
+
TEST(PaimonHybridReaderTest, DispatchesNativeThenJniSplitToMatchingReader) {
RuntimeProfile profile("test_profile");
RuntimeState state {TQueryOptions(), TQueryGlobals()};
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 0cc3e814082..46caf45aaee 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
@@ -314,19 +314,19 @@ public class PaimonJniScanner extends JniScanner {
if (reader != null) {
try {
reader.close();
+ reader = null;
} catch (IOException e) {
if (exception == null) {
exception = e;
} else {
exception.addSuppressed(e);
}
- } finally {
- reader = null;
}
}
if (ioManager != null) {
try {
ioManager.close();
+ ioManager = null;
} catch (Exception e) {
LOG.warn("Failed to close Paimon JNI IOManager, temp dirs:
{}", ioManagerTempDirs, e);
if (exception == null) {
@@ -334,8 +334,6 @@ public class PaimonJniScanner extends JniScanner {
} else {
exception.addSuppressed(e);
}
- } finally {
- ioManager = null;
}
}
} finally {
@@ -348,11 +346,8 @@ public class PaimonJniScanner extends JniScanner {
private void releaseRecordIterator() {
if (recordIterator != null) {
- try {
- recordIterator.releaseBatch();
- } finally {
- recordIterator = null;
- }
+ recordIterator.releaseBatch();
+ recordIterator = null;
}
}
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 6f893b38b1c..d43f251f58d 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
@@ -31,6 +31,7 @@ import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import java.io.File;
+import java.io.IOException;
import java.lang.reflect.Field;
import java.lang.reflect.Proxy;
import java.util.Arrays;
@@ -40,6 +41,7 @@ import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
public class PaimonJniScannerTest {
@Rule
@@ -190,6 +192,69 @@ public class PaimonJniScannerTest {
Assert.assertNull(recordIteratorField.get(scanner));
}
+ @Test
+ public void testFailedCloseRetainsResourcesForRetry() throws Exception {
+ PaimonJniScanner scanner = new PaimonJniScanner(128,
createBaseParams());
+ AtomicInteger iteratorCloseCalls = new AtomicInteger();
+ RecordReader.RecordIterator<InternalRow> recordIterator =
+ new RecordReader.RecordIterator<InternalRow>() {
+ @Override
+ public InternalRow next() {
+ return null;
+ }
+
+ @Override
+ public void releaseBatch() {
+ if (iteratorCloseCalls.incrementAndGet() == 1) {
+ throw new RuntimeException("injected iterator
close failure");
+ }
+ }
+ };
+ AtomicInteger readerCloseCalls = new AtomicInteger();
+ RecordReader<InternalRow> reader = new RecordReader<InternalRow>() {
+ @Override
+ public RecordIterator<InternalRow> readBatch() {
+ return null;
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (readerCloseCalls.incrementAndGet() == 1) {
+ throw new IOException("injected reader close failure");
+ }
+ }
+ };
+ RetryableIOManager ioManager = new RetryableIOManager();
+
+ Field recordIteratorField =
PaimonJniScanner.class.getDeclaredField("recordIterator");
+ recordIteratorField.setAccessible(true);
+ recordIteratorField.set(scanner, recordIterator);
+ Field readerField = PaimonJniScanner.class.getDeclaredField("reader");
+ readerField.setAccessible(true);
+ readerField.set(scanner, reader);
+ Field ioManagerField =
PaimonJniScanner.class.getDeclaredField("ioManager");
+ ioManagerField.setAccessible(true);
+ ioManagerField.set(scanner, ioManager);
+
+ try {
+ scanner.close();
+ Assert.fail("expected the first close to fail");
+ } catch (IOException expected) {
+ Assert.assertEquals("Failed to release Paimon record iterator",
expected.getMessage());
+ }
+ Assert.assertSame(recordIterator, recordIteratorField.get(scanner));
+ Assert.assertSame(reader, readerField.get(scanner));
+ Assert.assertSame(ioManager, ioManagerField.get(scanner));
+
+ scanner.close();
+ Assert.assertNull(recordIteratorField.get(scanner));
+ Assert.assertNull(readerField.get(scanner));
+ Assert.assertNull(ioManagerField.get(scanner));
+ Assert.assertEquals(2, iteratorCloseCalls.get());
+ Assert.assertEquals(2, readerCloseCalls.get());
+ Assert.assertEquals(2, ioManager.closeCalls.get());
+ }
+
private Map<String, String> createBaseParams() {
Map<String, String> params = new HashMap<>();
params.put("required_fields", "");
@@ -257,6 +322,21 @@ public class PaimonJniScannerTest {
}
}
+ public static class RetryableIOManager extends TestIOManager {
+ private final AtomicInteger closeCalls = new AtomicInteger();
+
+ public RetryableIOManager() {
+ super(new String[0]);
+ }
+
+ @Override
+ public void close() {
+ if (closeCalls.incrementAndGet() == 1) {
+ throw new RuntimeException("injected IO manager close
failure");
+ }
+ }
+ }
+
@Test
public void testGetFieldIndexMatchesMixedCaseColumns() {
Assert.assertEquals(1,
PaimonJniScanner.getFieldIndex(Arrays.asList("data", "mIxEd_COL", "PART"),
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
