From f4dd46c9839e89e245846a4a58308e8de5482540 Mon Sep 17 00:00:00 2001 From: Gabriel Date: Sun, 12 Jul 2026 22:14:39 +0800 Subject: [PATCH 1/5] [fix](be) Harden JNI reader split lifecycle ### What problem does this PR solve? JNI-backed table readers could violate scanner selection, cancellation, cleanup, EOF, adaptive batching, and rolling-upgrade contracts. Issue Number: N/A Related PR: N/A ### Release note Harden JNI table-reader lifecycle handling and compatibility for Paimon and Hudi scans. ### Check List - [x] Added unit tests for the fixes - [x] Added comments for compatibility and lifecycle invariants - [x] Passed 45 focused ASAN BE unit tests - [x] Passed repository format and diff checks --- be/src/exec/operator/file_scan_operator.cpp | 11 +- be/src/format_v2/jni/jni_table_reader.cpp | 94 ++++++++--- be/src/format_v2/jni/jni_table_reader.h | 17 +- be/src/format_v2/jni/paimon_jni_reader.cpp | 12 +- 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 | 21 +++ .../format_v2/jni/jni_table_reader_test.cpp | 147 ++++++++++++++++++ .../format_v2/jni/paimon_jni_reader_test.cpp | 36 +++++ be/test/format_v2/table/hudi_reader_test.cpp | 9 ++ .../format_v2/table/paimon_reader_test.cpp | 9 ++ 14 files changed, 394 insertions(+), 30 deletions(-) create mode 100644 be/test/format_v2/jni/jni_table_reader_test.cpp diff --git a/be/src/exec/operator/file_scan_operator.cpp b/be/src/exec/operator/file_scan_operator.cpp index 58aadfc87a6017..86f04c1dae18cc 100644 --- a/be/src/exec/operator/file_scan_operator.cpp +++ b/be/src/exec/operator/file_scan_operator.cpp @@ -116,10 +116,19 @@ 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"; + const bool uses_paimon_cpp_reader = + scan_params.__isset.table_format_params && + scan_params.table_format_params.table_format_type == "paimon" && + query_options.__isset.enable_paimon_cpp_reader && + query_options.enable_paimon_cpp_reader; + // PAIMON_CPP is selected per split by FE, while scanner selection only sees scan-level + // parameters. FileScannerV2 cannot dispatch that reader type, so retain the V1 path until the + // C++ Paimon reader is supported by the V2 hybrid reader. 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 && !is_transactional_hive && + !uses_paimon_cpp_reader; } Status FileScanLocalState::_init_scanners(std::list* scanners) { diff --git a/be/src/format_v2/jni/jni_table_reader.cpp b/be/src/format_v2/jni/jni_table_reader.cpp index 7cec22a7c5999f..b2056e47cfcb70 100644 --- a/be/src/format_v2/jni/jni_table_reader.cpp +++ b/be/src/format_v2/jni/jni_table_reader.cpp @@ -44,6 +44,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)); @@ -69,13 +72,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 @@ -287,9 +298,15 @@ Status JniTableReader::close() { if (_closed) { return Status::OK(); } - _closed = true; - RETURN_IF_ERROR(_close_jni_scanner()); - return TableReader::close(); + 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() { @@ -309,15 +326,23 @@ Status JniTableReader::_close_jni_scanner() { 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, @@ -329,11 +354,15 @@ Status JniTableReader::_close_jni_scanner() { _collect_jni_scanner_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()); + // 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); + } _reset_split_state(env); - return Status::OK(); + return cleanup_status; } void JniTableReader::_reset_split_state(JNIEnv* env) { @@ -342,7 +371,6 @@ 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(); @@ -371,14 +399,42 @@ 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(_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) { + 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(batch_size)) + .call(); +} + void JniTableReader::_prepare_jni_scanner_schema() { std::vector required_fields; std::vector 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 f9c854e1971fd0..ea8c353e0b0b6c 100644 --- a/be/src/format_v2/jni/jni_table_reader.h +++ b/be/src/format_v2/jni/jni_table_reader.h @@ -51,6 +51,17 @@ class JniTableReader : public TableReader { 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; + } + 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 +74,9 @@ class JniTableReader : public TableReader { 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); const std::vector& jni_columns() const { return _jni_columns; } TFileRangeDesc _current_range; @@ -77,13 +91,10 @@ class JniTableReader : public TableReader { 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* result); void _collect_jni_scanner_profile(JNIEnv* env); - Status _close_jni_scanner(); - std::map _scanner_params; std::vector _jni_columns; Block _jni_block_template; diff --git a/be/src/format_v2/jni/paimon_jni_reader.cpp b/be/src/format_v2/jni/paimon_jni_reader.cpp index ca1e602cdfedb9..47c0ef6c7bcac4 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* 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* 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/table/hudi_reader.cpp b/be/src/format_v2/table/hudi_reader.cpp index 3600b0ff706f08..f37d24bd888240 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 58f222c67abed9..e22c6bd866f061 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 +#include #include #include "format_v2/table_reader.h" @@ -61,6 +62,17 @@ class HudiHybridReader final : public format::TableReader { 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(); + _jni_reader = std::make_unique(); + } + std::pair 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 c2114968946e1b..236daac118a888 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 ae179e6135a9f1..fc94777b6dafb3 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 + #include "format_v2/table_reader.h" namespace doris { @@ -66,6 +68,7 @@ class PaimonHybridReader final : public format::TableReader { 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 @@ class PaimonHybridReader final : public format::TableReader { format::FileFormat* file_format) { return _to_file_format(range, file_format); } + void TEST_install_batch_size_children() { + _native_reader = std::make_unique(); + _jni_reader = std::make_unique(); + } + std::pair 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 da2cb351ff68ce..f5e1aea5ca5af2 100644 --- a/be/src/format_v2/table_reader.h +++ b/be/src/format_v2/table_reader.h @@ -167,13 +167,17 @@ class TableReader { // 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(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 9df8dcb54f4934..e7f27b6f7d423b 100644 --- a/be/test/exec/scan/file_scanner_v2_test.cpp +++ b/be/test/exec/scan/file_scanner_v2_test.cpp @@ -185,6 +185,27 @@ TEST(FileScannerV2Test, FileScanLocalStateSelectsV2ForSupportedQueriesOnly) { EXPECT_FALSE(FileScanLocalState::TEST_should_use_file_scanner_v2(query_options, false, params)); } +TEST(FileScannerV2Test, PaimonCppReaderForcesLegacyScanner) { + 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); + TTableFormatFileDesc table_format_params; + table_format_params.__set_table_format_type("paimon"); + params.__set_table_format_params(std::move(table_format_params)); + + EXPECT_FALSE(FileScanLocalState::TEST_should_use_file_scanner_v2(query_options, false, params)); + + // Other JNI table formats and Paimon without the C++ reader remain eligible for V2. + params.table_format_params.__set_table_format_type("hudi"); + EXPECT_TRUE(FileScanLocalState::TEST_should_use_file_scanner_v2(query_options, false, params)); + params.table_format_params.__set_table_format_type("paimon"); + query_options.__set_enable_paimon_cpp_reader(false); + EXPECT_TRUE(FileScanLocalState::TEST_should_use_file_scanner_v2(query_options, false, params)); +} + // 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 00000000000000..80cba23e8b5675 --- /dev/null +++ b/be/test/format_v2/jni/jni_table_reader_test.cpp @@ -0,0 +1,147 @@ +// 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 + +#include +#include +#include +#include + +#include "io/io_common.h" + +namespace doris::format { +namespace { + +class FakeJniTableReader final : public JniTableReader { +public: + int get_next_calls = 0; + int close_calls = 0; + std::vector propagated_batch_sizes; + std::vector close_results; + bool next_eof = false; + +protected: + std::string connector_class() const override { return "test/FakeJniScanner"; } + + Status build_scanner_params(std::map* 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 { + ++close_calls; + TEST_set_split_state(false, TEST_eof()); + if (static_cast(close_calls) <= close_results.size()) { + return close_results[close_calls - 1]; + } + return Status::OK(); + } + + Status _set_open_scanner_batch_size(size_t batch_size) override { + propagated_batch_sizes.push_back(batch_size); + return Status::OK(); + } +}; + +Status init_reader(FakeJniTableReader* reader, const std::shared_ptr& io_ctx) { + return reader->init({ + .projected_columns = {}, + .conjuncts = {}, + .format = FileFormat::JNI, + .scan_params = nullptr, + .io_ctx = io_ctx, + .runtime_state = nullptr, + .scanner_profile = nullptr, + }); +} + +TEST(JniTableReaderTest, CancellationStopsBeforeFetchingAnotherJavaBatch) { + auto io_ctx = std::make_shared(); + 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) { + FakeJniTableReader reader; + ASSERT_TRUE(init_reader(&reader, nullptr).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_EQ(reader.close_calls, 1); + + EXPECT_TRUE(reader.close().ok()); + EXPECT_TRUE(reader.TEST_closed()); + EXPECT_EQ(reader.close_calls, 2); +} + +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({33})); +} + +} // 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 08fd333abd3ada..592fdb54bd2e8a 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,41 @@ 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 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 params; + ASSERT_TRUE(build_params(&reader, range, ¶ms).ok()); + EXPECT_EQ(params["paimon.source"], "scan"); + EXPECT_EQ(params["hadoop.source"], "scan"); +} + } // 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 125183cd7a60c4..77cd75bc948ac8 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 f1c9196e733176..a56ebcc5e87151 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()}; From 52bda3f2c241f23986c5350bfbf636b9afa9687c Mon Sep 17 00:00:00 2001 From: Gabriel Date: Mon, 13 Jul 2026 09:23:29 +0800 Subject: [PATCH 2/5] [fix](be) Address JNI reader review feedback ### What problem does this PR solve? Issue Number: N/A Related PR: #65503 Problem Summary: Paimon C++ splits could still enter FileScannerV2 because scanner selection inspected a scan-level table-format descriptor that FE only sends per split. Failed JNI cleanup also discarded the Java object before a retry, and adaptive JNI sizing was applied after Paimon had already opened its physical reader. Use the scan-level Paimon predicate marker for routing, retain JNI state until cleanup succeeds, and seed adaptive batch sizing before split preparation opens JNI readers. ### Release note Ensure Paimon C++ scans use the supported scanner path, JNI cleanup remains retryable, and the first Paimon JNI batch honors adaptive sizing. ### Check List (For Author) - Test: Unit Test - Remote ASAN BE unit tests: FileScannerV2Test.PaimonCppReaderForcesLegacyScanner and JniTableReaderTest.* (6 tests passed) - Remote build-support/check-format.sh passed with clang-format 16 - Remote ASAN_UT changed objects compiled successfully - Remote clang-tidy analyzers completed for changed production files; final wrapper summary was not captured after the SSH output channel ended - Behavior changed: Yes, scanner routing, JNI cleanup retry state, and first-batch sizing are corrected - Does this need documentation: No --- be/src/exec/operator/file_scan_operator.cpp | 14 +++---- be/src/exec/scan/file_scanner_v2.cpp | 8 +++- be/src/format_v2/jni/jni_table_reader.cpp | 5 ++- be/src/format_v2/jni/jni_table_reader.h | 2 +- be/test/exec/scan/file_scanner_v2_test.cpp | 10 ++--- .../format_v2/jni/jni_table_reader_test.cpp | 39 +++++++++++++++++-- 6 files changed, 59 insertions(+), 19 deletions(-) diff --git a/be/src/exec/operator/file_scan_operator.cpp b/be/src/exec/operator/file_scan_operator.cpp index 86f04c1dae18cc..ecd0929d54713b 100644 --- a/be/src/exec/operator/file_scan_operator.cpp +++ b/be/src/exec/operator/file_scan_operator.cpp @@ -116,14 +116,12 @@ 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"; - const bool uses_paimon_cpp_reader = - scan_params.__isset.table_format_params && - scan_params.table_format_params.table_format_type == "paimon" && - query_options.__isset.enable_paimon_cpp_reader && - query_options.enable_paimon_cpp_reader; - // PAIMON_CPP is selected per split by FE, while scanner selection only sees scan-level - // parameters. FileScannerV2 cannot dispatch that reader type, so retain the V1 path until the - // C++ Paimon reader is supported by the V2 hybrid reader. + // FE stores Paimon's table-format descriptor per split, but paimon_predicate is a scan-level + // Paimon marker. PAIMON_CPP is also selected per split, and FileScannerV2 cannot dispatch that + // reader type, so retain the V1 path until the V2 hybrid reader supports it. + const bool uses_paimon_cpp_reader = scan_params.__isset.paimon_predicate && + query_options.__isset.enable_paimon_cpp_reader && + query_options.enable_paimon_cpp_reader; 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 && diff --git a/be/src/exec/scan/file_scanner_v2.cpp b/be/src/exec/scan/file_scanner_v2.cpp index 10b5f850ea36f7..c677bf8407e7a7 100644 --- a/be/src/exec/scan/file_scanner_v2.cpp +++ b/be/src/exec/scan/file_scanner_v2.cpp @@ -407,6 +407,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 partition_values; RETURN_IF_ERROR(_generate_partition_values(_current_range, &partition_values)); const auto status = @@ -422,7 +429,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; diff --git a/be/src/format_v2/jni/jni_table_reader.cpp b/be/src/format_v2/jni/jni_table_reader.cpp index b2056e47cfcb70..ce1b94294e4e19 100644 --- a/be/src/format_v2/jni/jni_table_reader.cpp +++ b/be/src/format_v2/jni/jni_table_reader.cpp @@ -361,7 +361,10 @@ Status JniTableReader::_close_jni_scanner() { if (cleanup_status.ok() && !java_close_status.ok()) { cleanup_status = std::move(java_close_status); } - _reset_split_state(env); + 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; } diff --git a/be/src/format_v2/jni/jni_table_reader.h b/be/src/format_v2/jni/jni_table_reader.h index ea8c353e0b0b6c..2b73c144e04674 100644 --- a/be/src/format_v2/jni/jni_table_reader.h +++ b/be/src/format_v2/jni/jni_table_reader.h @@ -77,6 +77,7 @@ class JniTableReader : public TableReader { 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 Status _open_jni_scanner(); const std::vector& jni_columns() const { return _jni_columns; } TFileRangeDesc _current_range; @@ -85,7 +86,6 @@ class JniTableReader : public TableReader { 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); diff --git a/be/test/exec/scan/file_scanner_v2_test.cpp b/be/test/exec/scan/file_scanner_v2_test.cpp index e7f27b6f7d423b..ef7a87d6d0aa60 100644 --- a/be/test/exec/scan/file_scanner_v2_test.cpp +++ b/be/test/exec/scan/file_scanner_v2_test.cpp @@ -192,16 +192,16 @@ TEST(FileScannerV2Test, PaimonCppReaderForcesLegacyScanner) { TFileScanRangeParams params; params.__set_format_type(TFileFormatType::FORMAT_JNI); - TTableFormatFileDesc table_format_params; - table_format_params.__set_table_format_type("paimon"); - params.__set_table_format_params(std::move(table_format_params)); + // Match FE-generated Paimon params: the table-format descriptor is stored per range, while + // paimon_predicate is stored once at scan level. + params.__set_paimon_predicate("serialized-predicate"); EXPECT_FALSE(FileScanLocalState::TEST_should_use_file_scanner_v2(query_options, false, params)); // Other JNI table formats and Paimon without the C++ reader remain eligible for V2. - params.table_format_params.__set_table_format_type("hudi"); + params.__isset.paimon_predicate = false; EXPECT_TRUE(FileScanLocalState::TEST_should_use_file_scanner_v2(query_options, false, params)); - params.table_format_params.__set_table_format_type("paimon"); + params.__set_paimon_predicate("serialized-predicate"); query_options.__set_enable_paimon_cpp_reader(false); EXPECT_TRUE(FileScanLocalState::TEST_should_use_file_scanner_v2(query_options, false, params)); } diff --git a/be/test/format_v2/jni/jni_table_reader_test.cpp b/be/test/format_v2/jni/jni_table_reader_test.cpp index 80cba23e8b5675..8162ec6bc5ddea 100644 --- a/be/test/format_v2/jni/jni_table_reader_test.cpp +++ b/be/test/format_v2/jni/jni_table_reader_test.cpp @@ -33,6 +33,7 @@ class FakeJniTableReader final : public JniTableReader { public: int get_next_calls = 0; int close_calls = 0; + std::vector open_batch_sizes; std::vector propagated_batch_sizes; std::vector close_results; bool next_eof = false; @@ -53,18 +54,30 @@ class FakeJniTableReader final : public JniTableReader { } Status _close_jni_scanner() override { + if (!TEST_scanner_opened()) { + return Status::OK(); + } ++close_calls; - TEST_set_split_state(false, TEST_eof()); + Status status = Status::OK(); if (static_cast(close_calls) <= close_results.size()) { - return close_results[close_calls - 1]; + status = close_results[close_calls - 1]; } - return Status::OK(); + 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_ctx) { @@ -122,6 +135,7 @@ TEST(JniTableReaderTest, FailedCloseCanBeRetried) { EXPECT_FALSE(reader.close().ok()); EXPECT_FALSE(reader.TEST_closed()); + EXPECT_TRUE(reader.TEST_scanner_opened()); EXPECT_EQ(reader.close_calls, 1); EXPECT_TRUE(reader.close().ok()); @@ -143,5 +157,24 @@ TEST(JniTableReaderTest, AdaptiveBatchSizeUpdatesAnOpenJavaScanner) { EXPECT_EQ(reader.propagated_batch_sizes, std::vector({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({32})); + EXPECT_TRUE(reader.TEST_scanner_opened()); +} + } // namespace } // namespace doris::format From 01a6b19a78f8fd342b6850a2889e70c1fd9cdba2 Mon Sep 17 00:00:00 2001 From: Gabriel Date: Mon, 13 Jul 2026 10:57:08 +0800 Subject: [PATCH 3/5] [fix](be) Address JNI scanner review feedback ### What problem does this PR solve? Issue Number: None Related PR: #65503 Problem Summary: FileScannerV2 selection could still route rolling-upgrade Paimon CPP JNI splits into an unsupported V2 reader, and a failed table-reader close left the scanner marked closed so retained JNI cleanup state could not be retried. Conservatively keep JNI scans on V1 while the Paimon C++ reader option is enabled, and roll back the scanner close guard when table-reader cleanup fails. ### Release note Harden Paimon CPP scanner selection and retry JNI cleanup after close failures. ### Check List (For Author) - Test: Unit Test - FileScannerV2Test.* and JniTableReaderTest.*: 20 tests passed under ASAN BE UT - Behavior changed: Yes. Unsupported Paimon CPP JNI splits stay on V1, and failed V2 table-reader cleanup can be retried. - Does this need documentation: No --- be/src/exec/operator/file_scan_operator.cpp | 15 +++--- be/src/exec/scan/file_scanner_v2.cpp | 12 ++++- be/src/exec/scan/file_scanner_v2.h | 2 + be/test/exec/scan/file_scanner_v2_test.cpp | 57 ++++++++++++++++++--- 4 files changed, 70 insertions(+), 16 deletions(-) diff --git a/be/src/exec/operator/file_scan_operator.cpp b/be/src/exec/operator/file_scan_operator.cpp index ecd0929d54713b..a570d279c4384f 100644 --- a/be/src/exec/operator/file_scan_operator.cpp +++ b/be/src/exec/operator/file_scan_operator.cpp @@ -116,17 +116,18 @@ 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"; - // FE stores Paimon's table-format descriptor per split, but paimon_predicate is a scan-level - // Paimon marker. PAIMON_CPP is also selected per split, and FileScannerV2 cannot dispatch that - // reader type, so retain the V1 path until the V2 hybrid reader supports it. - const bool uses_paimon_cpp_reader = scan_params.__isset.paimon_predicate && - query_options.__isset.enable_paimon_cpp_reader && - query_options.enable_paimon_cpp_reader; + // PAIMON_CPP is selected per split, but this scan-level selector cannot inspect split-level + // table_format_params. Older FEs may also omit the scan-level paimon_predicate marker. When the + // C++ reader option is enabled, conservatively keep JNI scans on V1 until FileScannerV2 can + // dispatch PAIMON_CPP ranges itself. + const bool may_use_paimon_cpp_reader = scan_params.format_type == TFileFormatType::FORMAT_JNI && + query_options.__isset.enable_paimon_cpp_reader && + query_options.enable_paimon_cpp_reader; 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 && - !uses_paimon_cpp_reader; + !may_use_paimon_cpp_reader; } Status FileScanLocalState::_init_scanners(std::list* scanners) { diff --git a/be/src/exec/scan/file_scanner_v2.cpp b/be/src/exec/scan/file_scanner_v2.cpp index c677bf8407e7a7..893be237e02c68 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 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); @@ -824,7 +828,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 66242b610f4373..2bddc5d5e69e6c 100644 --- a/be/src/exec/scan/file_scanner_v2.h +++ b/be/src/exec/scan/file_scanner_v2.h @@ -65,6 +65,8 @@ class FileScannerV2 final : public Scanner { static bool is_supported(const TFileScanRangeParams& params, const TFileRangeDesc& range); #ifdef BE_TEST + FileScannerV2(RuntimeState* state, RuntimeProfile* profile, + std::unique_ptr 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/test/exec/scan/file_scanner_v2_test.cpp b/be/test/exec/scan/file_scanner_v2_test.cpp index ef7a87d6d0aa60..c93996c040877a 100644 --- a/be/test/exec/scan/file_scanner_v2_test.cpp +++ b/be/test/exec/scan/file_scanner_v2_test.cpp @@ -59,6 +59,35 @@ 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; +} + +struct RetryableCloseState { + int close_calls = 0; +}; + +class RetryableCloseTableReader final : public format::TableReader { +public: + explicit RetryableCloseTableReader(std::shared_ptr 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 _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); } @@ -192,20 +221,32 @@ TEST(FileScannerV2Test, PaimonCppReaderForcesLegacyScanner) { TFileScanRangeParams params; params.__set_format_type(TFileFormatType::FORMAT_JNI); - // Match FE-generated Paimon params: the table-format descriptor is stored per range, while - // paimon_predicate is stored once at scan level. - params.__set_paimon_predicate("serialized-predicate"); - + // Rolling upgrades may carry the only Paimon marker and reader type on each split. Since the + // scan-level selector cannot inspect that split yet, the C++ reader option conservatively keeps + // JNI scans on V1 even without a scan-level paimon_predicate. EXPECT_FALSE(FileScanLocalState::TEST_should_use_file_scanner_v2(query_options, false, params)); + EXPECT_FALSE(FileScannerV2::is_supported(params, paimon_cpp_jni_range())); - // Other JNI table formats and Paimon without the C++ reader remain eligible for V2. - params.__isset.paimon_predicate = false; - EXPECT_TRUE(FileScanLocalState::TEST_should_use_file_scanner_v2(query_options, false, params)); - params.__set_paimon_predicate("serialized-predicate"); + // Disabling the C++ reader restores V2 eligibility for supported JNI implementations. query_options.__set_enable_paimon_cpp_reader(false); EXPECT_TRUE(FileScanLocalState::TEST_should_use_file_scanner_v2(query_options, false, params)); } +TEST(FileScannerV2Test, FailedTableReaderCloseCanBeRetriedThroughScanner) { + RuntimeState state {TQueryOptions(), TQueryGlobals()}; + RuntimeProfile profile("file_scanner_v2_close_retry"); + auto close_state = std::make_shared(); + FileScannerV2 scanner(&state, &profile, + std::make_unique(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) { From 68655aebade5fe2cf92f8a045c125017555ffe31 Mon Sep 17 00:00:00 2001 From: Gabriel Date: Mon, 13 Jul 2026 11:54:50 +0800 Subject: [PATCH 4/5] [fix](be) Complete JNI scanner review fixes ### What problem does this PR solve? Issue Number: None Related PR: #65503 Problem Summary: Legacy Paimon JNI splits without reader_type could still enter FileScannerV2, adaptive batch-size updates after open did not resize Paimon physical readers, and failed Java cleanup cleared resource references before a retry. Keep JNI compatibility shapes on the legacy scanner, preserve the initial Paimon physical batch size after open, and retain each Java resource until its cleanup succeeds. ### Release note Harden legacy JNI selection, Paimon batch-size handling, and retryable Java cleanup. ### Check List (For Author) - Test: Unit Test - FileScannerV2Test.*, JniTableReaderTest.*, and PaimonJniReaderTest.*: 30 tests passed under ASAN BE UT - PaimonJniScannerTest: 10 tests passed; checkstyle reported 0 violations - clang-format and git diff checks passed - Behavior changed: Yes. Legacy JNI compatibility splits stay on V1, Paimon ignores unsupported post-open physical batch resizing, and failed Java cleanup remains retryable. - Does this need documentation: No --- be/src/exec/operator/file_scan_operator.cpp | 14 ++-- be/src/format_v2/jni/jni_table_reader.cpp | 5 ++ be/src/format_v2/jni/jni_table_reader.h | 1 + be/src/format_v2/jni/paimon_jni_reader.h | 1 + be/test/exec/scan/file_scanner_v2_test.cpp | 21 +++-- .../format_v2/jni/paimon_jni_reader_test.cpp | 12 +++ .../apache/doris/paimon/PaimonJniScanner.java | 13 +-- .../doris/paimon/PaimonJniScannerTest.java | 80 +++++++++++++++++++ 8 files changed, 124 insertions(+), 23 deletions(-) diff --git a/be/src/exec/operator/file_scan_operator.cpp b/be/src/exec/operator/file_scan_operator.cpp index a570d279c4384f..8121590ebf04f4 100644 --- a/be/src/exec/operator/file_scan_operator.cpp +++ b/be/src/exec/operator/file_scan_operator.cpp @@ -116,18 +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"; - // PAIMON_CPP is selected per split, but this scan-level selector cannot inspect split-level - // table_format_params. Older FEs may also omit the scan-level paimon_predicate marker. When the - // C++ reader option is enabled, conservatively keep JNI scans on V1 until FileScannerV2 can - // dispatch PAIMON_CPP ranges itself. - const bool may_use_paimon_cpp_reader = scan_params.format_type == TFileFormatType::FORMAT_JNI && - query_options.__isset.enable_paimon_cpp_reader && - query_options.enable_paimon_cpp_reader; + // 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 && - !may_use_paimon_cpp_reader; + scan_params.format_type != TFileFormatType::FORMAT_LANCE && + scan_params.format_type != TFileFormatType::FORMAT_JNI && !is_transactional_hive; } Status FileScanLocalState::_init_scanners(std::list* scanners) { diff --git a/be/src/format_v2/jni/jni_table_reader.cpp b/be/src/format_v2/jni/jni_table_reader.cpp index ce1b94294e4e19..bb7738ab098529 100644 --- a/be/src/format_v2/jni/jni_table_reader.cpp +++ b/be/src/format_v2/jni/jni_table_reader.cpp @@ -418,6 +418,11 @@ Status JniTableReader::_open_jni_scanner() { } 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; diff --git a/be/src/format_v2/jni/jni_table_reader.h b/be/src/format_v2/jni/jni_table_reader.h index 2b73c144e04674..c19bf7c101bbdf 100644 --- a/be/src/format_v2/jni/jni_table_reader.h +++ b/be/src/format_v2/jni/jni_table_reader.h @@ -77,6 +77,7 @@ class JniTableReader : public TableReader { 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(); const std::vector& jni_columns() const { return _jni_columns; } TFileRangeDesc _current_range; diff --git a/be/src/format_v2/jni/paimon_jni_reader.h b/be/src/format_v2/jni/paimon_jni_reader.h index 307880815654fd..26767458b427db 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 @@ class PaimonJniReader final : public format::JniTableReader { std::string connector_class() const override; Status validate_scan_range(const TFileRangeDesc& range) const override; Status build_scanner_params(std::map* 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& store_paths); diff --git a/be/test/exec/scan/file_scanner_v2_test.cpp b/be/test/exec/scan/file_scanner_v2_test.cpp index c93996c040877a..2137082bb10fdd 100644 --- a/be/test/exec/scan/file_scanner_v2_test.cpp +++ b/be/test/exec/scan/file_scanner_v2_test.cpp @@ -67,6 +67,15 @@ TFileRangeDesc paimon_cpp_jni_range() { 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; }; @@ -214,7 +223,7 @@ TEST(FileScannerV2Test, FileScanLocalStateSelectsV2ForSupportedQueriesOnly) { EXPECT_FALSE(FileScanLocalState::TEST_should_use_file_scanner_v2(query_options, false, params)); } -TEST(FileScannerV2Test, PaimonCppReaderForcesLegacyScanner) { +TEST(FileScannerV2Test, JniCompatibilityShapesForceLegacyScanner) { TQueryOptions query_options; query_options.__set_enable_file_scanner_v2(true); query_options.__set_enable_paimon_cpp_reader(true); @@ -222,14 +231,16 @@ TEST(FileScannerV2Test, PaimonCppReaderForcesLegacyScanner) { 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, the C++ reader option conservatively keeps - // JNI scans on V1 even without a scan-level paimon_predicate. + // 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())); - // Disabling the C++ reader restores V2 eligibility for supported JNI implementations. + // 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_TRUE(FileScanLocalState::TEST_should_use_file_scanner_v2(query_options, false, params)); + 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) { 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 592fdb54bd2e8a..570977a5f7eb4e 100644 --- a/be/test/format_v2/jni/paimon_jni_reader_test.cpp +++ b/be/test/format_v2/jni/paimon_jni_reader_test.cpp @@ -159,5 +159,17 @@ TEST(PaimonJniReaderTest, ScanLevelOptionsOverrideLegacySplitFallbacks) { 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/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 0cc3e814082711..46caf45aaeee8c 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 void close() throws IOException { 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 void close() throws IOException { } else { exception.addSuppressed(e); } - } finally { - ioManager = null; } } } finally { @@ -348,11 +346,8 @@ public void close() throws IOException { 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 6f893b38b1cf9f..d43f251f58deb1 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.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.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 void releaseBatch() { Assert.assertNull(recordIteratorField.get(scanner)); } + @Test + public void testFailedCloseRetainsResourcesForRetry() throws Exception { + PaimonJniScanner scanner = new PaimonJniScanner(128, createBaseParams()); + AtomicInteger iteratorCloseCalls = new AtomicInteger(); + RecordReader.RecordIterator recordIterator = + new RecordReader.RecordIterator() { + @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 reader = new RecordReader() { + @Override + public RecordIterator 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 createBaseParams() { Map params = new HashMap<>(); params.put("required_fields", ""); @@ -257,6 +322,21 @@ public void close() { } } + 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"), From c05c7b5389214a58609b9a7be95c9f3eaa897ade Mon Sep 17 00:00:00 2001 From: Gabriel Date: Mon, 13 Jul 2026 12:56:05 +0800 Subject: [PATCH 5/5] [fix](be) Avoid duplicate JNI close profiles ### What problem does this PR solve? Issue Number: None Related PR: #65503 Problem Summary: A retryable JNI scanner close published the same split profile on every cleanup attempt because the Java scanner and watcher values must remain intact after a failed close. Reserve profile publication per split so close retries only repeat resource cleanup, then reset that publication state when the split is successfully released. ### Release note Keep JNI split profile counters idempotent across close retries. ### Check List (For Author) - Test: Unit Test - FileScannerV2Test.*, JniTableReaderTest.*, and PaimonJniReaderTest.*: 30 tests passed under ASAN BE UT - FailedCloseCanBeRetried asserts the close profile counter is unchanged after a fail-then-success retry - check-format and git diff checks passed - Behavior changed: Yes. Retried JNI cleanup no longer publishes the same split profile more than once. - Does this need documentation: No --- be/src/format_v2/jni/jni_table_reader.cpp | 63 ++++++++++++------- be/src/format_v2/jni/jni_table_reader.h | 6 ++ .../format_v2/jni/jni_table_reader_test.cpp | 17 ++++- 3 files changed, 61 insertions(+), 25 deletions(-) diff --git a/be/src/format_v2/jni/jni_table_reader.cpp b/be/src/format_v2/jni/jni_table_reader.cpp index bb7738ab098529..e8a6cc376e9e55 100644 --- a/be/src/format_v2/jni/jni_table_reader.cpp +++ b/be/src/format_v2/jni/jni_table_reader.cpp @@ -294,33 +294,21 @@ 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; } - 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; + _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); @@ -352,6 +340,36 @@ 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. Java close must still run if that @@ -380,6 +398,7 @@ void JniTableReader::_reset_split_state(JNIEnv* env) { _jni_scanner_open_watcher = 0; _java_scan_watcher = 0; _fill_block_watcher = 0; + _split_profile_published = false; } Status JniTableReader::_open_jni_scanner() { diff --git a/be/src/format_v2/jni/jni_table_reader.h b/be/src/format_v2/jni/jni_table_reader.h index c19bf7c101bbdf..24f84ca8d62756 100644 --- a/be/src/format_v2/jni/jni_table_reader.h +++ b/be/src/format_v2/jni/jni_table_reader.h @@ -57,6 +57,9 @@ class JniTableReader : public TableReader { 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; } @@ -79,6 +82,7 @@ class JniTableReader : public TableReader { 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& jni_columns() const { return _jni_columns; } TFileRangeDesc _current_range; @@ -95,6 +99,7 @@ class JniTableReader : public TableReader { Status _fill_jni_block(JniDataBridge::TableMetaAddress& table_meta, size_t num_rows); Status _get_statistics(JNIEnv* env, std::map* result); void _collect_jni_scanner_profile(JNIEnv* env); + void _publish_split_profile(JNIEnv* env); std::map _scanner_params; std::vector _jni_columns; @@ -103,6 +108,7 @@ class JniTableReader : public TableReader { 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/test/format_v2/jni/jni_table_reader_test.cpp b/be/test/format_v2/jni/jni_table_reader_test.cpp index 8162ec6bc5ddea..1c5788b1df725c 100644 --- a/be/test/format_v2/jni/jni_table_reader_test.cpp +++ b/be/test/format_v2/jni/jni_table_reader_test.cpp @@ -33,6 +33,8 @@ 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 open_batch_sizes; std::vector propagated_batch_sizes; std::vector close_results; @@ -58,6 +60,9 @@ class FakeJniTableReader final : public JniTableReader { 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(close_calls) <= close_results.size()) { status = close_results[close_calls - 1]; @@ -80,7 +85,8 @@ class FakeJniTableReader final : public JniTableReader { } }; -Status init_reader(FakeJniTableReader* reader, const std::shared_ptr& io_ctx) { +Status init_reader(FakeJniTableReader* reader, const std::shared_ptr& io_ctx, + RuntimeProfile* scanner_profile = nullptr) { return reader->init({ .projected_columns = {}, .conjuncts = {}, @@ -88,7 +94,7 @@ Status init_reader(FakeJniTableReader* reader, const std::shared_ptrvalue(), 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) {