diff --git a/be/src/exec/operator/file_scan_operator.cpp b/be/src/exec/operator/file_scan_operator.cpp index 58aadfc87a6017..8121590ebf04f4 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* scanners) { diff --git a/be/src/exec/scan/file_scanner_v2.cpp b/be/src/exec/scan/file_scanner_v2.cpp index 10b5f850ea36f7..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); @@ -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 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; @@ -818,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/src/format_v2/jni/jni_table_reader.cpp b/be/src/format_v2/jni/jni_table_reader.cpp index 7cec22a7c5999f..e8a6cc376e9e55 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 @@ -283,41 +294,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, @@ -327,13 +340,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) { @@ -342,13 +392,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() { @@ -371,14 +421,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(_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(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..24f84ca8d62756 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 @@ 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; + 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 @@ 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); + 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; @@ -71,18 +91,15 @@ 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); 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(); + void _publish_split_profile(JNIEnv* env); std::map _scanner_params; std::vector _jni_columns; @@ -91,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/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/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/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..2137082bb10fdd 100644 --- a/be/test/exec/scan/file_scanner_v2_test.cpp +++ b/be/test/exec/scan/file_scanner_v2_test.cpp @@ -59,6 +59,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 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); } @@ -185,6 +223,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(); + 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) { 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..1c5788b1df725c --- /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 + +#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; + 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; + 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 { + 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(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_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(); + 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({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 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..570977a5f7eb4e 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 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"); +} + +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 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()}; 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"),