[fix](be) Harden JNI reader split lifecycle

be/src/exec/operator/file_scan_operator.cpp

@@ -116,10 +116,14 @@ bool FileScanLocalState::_should_use_file_scanner_v2(const TQueryOptions& query_
     const bool is_transactional_hive =
             scan_params.__isset.table_format_params &&
             scan_params.table_format_params.table_format_type == "transactional_hive";
+    // JNI reader selection is stored per split, but this scan-level selector cannot inspect the
+    // split yet. Older FEs may omit both the scan-level Paimon marker and split-level reader_type,
+    // so keep JNI scans on V1 until scanner selection can distinguish every compatibility shape.
     return query_options.__isset.enable_file_scanner_v2 && query_options.enable_file_scanner_v2 &&
            !is_load && scan_params.format_type != TFileFormatType::FORMAT_WAL &&
            scan_params.format_type != TFileFormatType::FORMAT_ES_HTTP &&
-           scan_params.format_type != TFileFormatType::FORMAT_LANCE && !is_transactional_hive;
+           scan_params.format_type != TFileFormatType::FORMAT_LANCE &&
+           scan_params.format_type != TFileFormatType::FORMAT_JNI && !is_transactional_hive;
 }
 
 Status FileScanLocalState::_init_scanners(std::list<ScannerSPtr>* scanners) {

be/src/exec/scan/file_scanner_v2.cpp

@@ -212,6 +212,10 @@ Status rewrite_slot_refs_to_global_index(
 } // namespace
 
 #ifdef BE_TEST
+FileScannerV2::FileScannerV2(RuntimeState* state, RuntimeProfile* profile,
+                             std::unique_ptr<format::TableReader> table_reader)
+        : Scanner(state, profile), _table_reader(std::move(table_reader)) {}
+
 Status FileScannerV2::TEST_validate_scan_range(const TFileScanRangeParams& params,
                                                const TFileRangeDesc& range) {
     return _validate_scan_range(params, range);
@@ -407,6 +411,13 @@ Status FileScannerV2::_prepare_next_split(bool* eos) {
         DORIS_CHECK(_table_reader != nullptr);
         _current_range_path = _current_range.path;
 
+        const auto format_type = get_range_format_type(*_params, _current_range);
+        _init_adaptive_batch_size_state(format_type);
+        if (_should_run_adaptive_batch_size()) {
+            // JNI readers open eagerly in prepare_split(). Seed the probe size first so readers
+            // such as Paimon also use it for their first physical read batch.
+            _table_reader->set_batch_size(_predict_reader_batch_rows());
+        }
         std::map<std::string, Field> partition_values;
         RETURN_IF_ERROR(_generate_partition_values(_current_range, &partition_values));
         const auto status =
@@ -422,7 +433,6 @@ Status FileScannerV2::_prepare_next_split(bool* eos) {
             _state->update_num_finished_scan_range(1);
             continue;
         }
-        _init_adaptive_batch_size_state(get_range_format_type(*_params, _current_range));
         COUNTER_UPDATE(_file_counter, 1);
         _has_prepared_split = true;
         *eos = false;
@@ -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();
     }

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<format::TableReader> table_reader);
     static Status TEST_validate_scan_range(const TFileScanRangeParams& params,
                                            const TFileRangeDesc& range);
     static Status TEST_to_file_format(TFileFormatType::type format_type,

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<int>(_batch_size)));
-    // call open() method in JAVA side.
-    RETURN_IF_ERROR(_jni_scanner_obj.call_void_method(env, _jni_scanner_open).call());
-    RETURN_ERROR_IF_EXC(env);
-
+    // Once the Java object exists, close it even if open() fails partway through initialization.
+    // Connector implementations may already own streams, off-heap tables, or JDBC connections.
     _scanner_opened = true;
+    // call open() method in JAVA side.
+    const auto open_status = _jni_scanner_obj.call_void_method(env, _jni_scanner_open).call();
+    if (!open_status.ok()) {
+        const auto close_status = _close_jni_scanner();
+        if (!close_status.ok()) {
+            LOG(WARNING) << "failed to clean up JNI scanner after open failure: " << close_status;
+        }
+        return open_status;
+    }
     return Status::OK();
 }
 
+void JniTableReader::set_batch_size(size_t batch_size) {
+    if (_scanner_opened && !supports_batch_size_update_after_open()) {
+        // Some connectors bake the constructor batch size into an already-open physical reader.
+        // Keep C++ and Java on that initial size instead of pretending a later resize took effect.
+        return;
+    }
+    TableReader::set_batch_size(batch_size);
+    if (!_scanner_opened) {
+        return;
+    }
+    const auto status = _set_open_scanner_batch_size(_batch_size);
+    if (!status.ok()) {
+        // Adaptive batch sizing is an optimization. Keep the scanner usable with its previous
+        // size if Java rejects a mid-split update, but surface the failure for diagnosis.
+        LOG(WARNING) << "failed to update JNI scanner batch size: " << status;
+    }
+}
+
+Status JniTableReader::_set_open_scanner_batch_size(size_t batch_size) {
+    JNIEnv* env = nullptr;
+    RETURN_IF_ERROR(Jni::Env::Get(&env));
+    return _jni_scanner_obj.call_void_method(env, _jni_scanner_set_batch_size)
+            .with_arg(cast_set<int>(batch_size))
+            .call();
+}
+
 void JniTableReader::_prepare_jni_scanner_schema() {
     std::vector<std::string> required_fields;
     std::vector<std::string> column_types;

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<JniColumn>& 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<std::string, std::string>* result);
     void _collect_jni_scanner_profile(JNIEnv* env);
-
-    Status _close_jni_scanner();
+    void _publish_split_profile(JNIEnv* env);
 
     std::map<std::string, std::string> _scanner_params;
     std::vector<JniColumn> _jni_columns;
@@ -91,6 +108,7 @@ 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;

be/src/format_v2/jni/paimon_jni_reader.cpp

@@ -99,6 +99,11 @@ Status PaimonJniReader::build_scanner_params(std::map<std::string, std::string>*
         for (const auto& kv : _scan_params->paimon_options) {
             (*params)[std::string(PAIMON_OPTION_PREFIX) + kv.first] = kv.second;
         }
+    } else if (paimon_params.__isset.paimon_options) {
+        // Rolling upgrades can pair this BE with an older FE that only sends options per split.
+        for (const auto& kv : paimon_params.paimon_options) {
+            (*params)[std::string(PAIMON_OPTION_PREFIX) + kv.first] = kv.second;
+        }
     }
     const std::string enable_io_manager_key =
             std::string(PAIMON_OPTION_PREFIX) + std::string(DORIS_ENABLE_JNI_IO_MANAGER);
@@ -118,10 +123,13 @@ Status PaimonJniReader::build_scanner_params(std::map<std::string, std::string>*
         for (const auto& kv : _scan_params->properties) {
             (*params)[std::string(HADOOP_OPTION_PREFIX) + kv.first] = kv.second;
         }
+    } else if (paimon_params.__isset.hadoop_conf) {
+        for (const auto& kv : paimon_params.hadoop_conf) {
+            (*params)[std::string(HADOOP_OPTION_PREFIX) + kv.first] = kv.second;
+        }
     }
     // TODO: Remove legacy split-level paimon_predicate, paimon_options and hadoop_conf from thrift
-    // after all readers stop using them. Predicate keeps the split-level fallback for rolling
-    // upgrade compatibility with old FE paths that did not send scan-level paimon_predicate.
+    // after the minimum supported FE always sends their scan-level replacements.
     return Status::OK();
 }
 

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<std::string, std::string>* params) const override;
+    bool supports_batch_size_update_after_open() const override { return false; }
 
 private:
     static std::string build_default_io_manager_tmp_dirs(const std::vector<StorePath>& store_paths);