diff --git a/be/src/format_v2/jni/jni_table_reader.cpp b/be/src/format_v2/jni/jni_table_reader.cpp
index deefe26837eeac..dd0dfacffd6c65 100644
--- a/be/src/format_v2/jni/jni_table_reader.cpp
+++ b/be/src/format_v2/jni/jni_table_reader.cpp
@@ -48,13 +48,12 @@ Status JniTableReader::prepare_split(const SplitReadOptions& options) {
     RETURN_IF_ERROR(validate_scan_range(options.current_range));
     RETURN_IF_ERROR(TableReader::prepare_split(options));
     DORIS_CHECK(!_closed);
-    DORIS_CHECK(!_scanner_opened);
     if (_is_table_level_count_active()) {
         return Status::OK();
     }
     // Subclasses populate split-specific scanner params before calling this method, so the Java
     // scanner can be opened here instead of being lazily opened by the first get_block() call.
-    return _open_jni_scanner();
+    return open_jni_scanner_for_split();
 }
 
 Status JniTableReader::get_block(Block* output_block, bool* eos) {
@@ -79,7 +78,7 @@ Status JniTableReader::get_block(Block* output_block, bool* eos) {
         RETURN_IF_ERROR(_get_next_jni_block(&current_rows, &current_eof));
         if (current_eof) {
             _eof = true;
-            RETURN_IF_ERROR(_close_jni_scanner());
+            RETURN_IF_ERROR(close_jni_scanner_for_split());
             *eos = true;
             return Status::OK();
         }
@@ -280,11 +279,11 @@ Status JniTableReader::close() {
         return Status::OK();
     }
     _closed = true;
-    RETURN_IF_ERROR(_close_jni_scanner());
+    RETURN_IF_ERROR(JniTableReader::close_jni_scanner_for_split());
     return TableReader::close();
 }
 
-Status JniTableReader::_close_jni_scanner() {
+Status JniTableReader::close_jni_scanner_for_split() {
     if (!_scanner_opened) {
         JNIEnv* env = nullptr;
         if (!_jni_scanner_obj.uninitialized()) {
@@ -343,7 +342,8 @@ void JniTableReader::_reset_split_state(JNIEnv* env) {
     _fill_block_watcher = 0;
 }
 
-Status JniTableReader::_open_jni_scanner() {
+Status JniTableReader::open_jni_scanner_for_split() {
+    DORIS_CHECK(!_scanner_opened);
     // subclasses build map<string,string> _scanner_params to JAVA side
     RETURN_IF_ERROR(build_scanner_params(&_scanner_params));
     // subclasses build _jni_columns info to JAVA side, including column name and column type
@@ -424,6 +424,10 @@ Status JniTableReader::_register_jni_class_functions_once(JNIEnv* env) {
                                                 &_jni_scanner_get_statistics));
     RETURN_IF_ERROR(
             _jni_scanner_cls.get_method(env, "setBatchSize", "(I)V", &_jni_scanner_set_batch_size));
+    RETURN_IF_ERROR(_jni_scanner_cls.get_method(env, "prepareForSplit", "(Ljava/util/Map;)V",
+                                                &_jni_scanner_prepare_for_split));
+    RETURN_IF_ERROR(_jni_scanner_cls.get_method(env, "resetCurrentSplit", "()V",
+                                                &_jni_scanner_reset_current_split));
     return Status::OK();
 }
 
diff --git a/be/src/format_v2/jni/jni_table_reader.h b/be/src/format_v2/jni/jni_table_reader.h
index cfe1eb32c69ee4..909ebda9117c5e 100644
--- a/be/src/format_v2/jni/jni_table_reader.h
+++ b/be/src/format_v2/jni/jni_table_reader.h
@@ -60,6 +60,17 @@ class JniTableReader : public TableReader {
     // Subclasses can override this method when Java transfer types differ from output types.
     virtual Status build_jni_columns(std::vector<JniColumn>* columns) const;
     virtual Status finalize_jni_block(Block* jni_block, Block* output_block, size_t* rows);
+    virtual Status open_jni_scanner_for_split();
+    virtual Status close_jni_scanner_for_split();
+    void reset_jni_eof() { _eof = false; }
+    bool jni_scanner_opened() const { return _scanner_opened; }
+    Jni::GlobalObject& jni_scanner_obj() { return _jni_scanner_obj; }
+    const Jni::MethodId& jni_scanner_prepare_for_split() const {
+        return _jni_scanner_prepare_for_split;
+    }
+    const Jni::MethodId& jni_scanner_reset_current_split() const {
+        return _jni_scanner_reset_current_split;
+    }
     // used for profile
     virtual int64_t self_split_weight() const;
     const std::vector<JniColumn>& jni_columns() const { return _jni_columns; }
@@ -69,8 +80,6 @@ class JniTableReader : public TableReader {
     // init
     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);
@@ -81,8 +90,6 @@ class JniTableReader : public TableReader {
     Status _get_statistics(JNIEnv* env, std::map<std::string, std::string>* result);
     void _collect_jni_scanner_profile(JNIEnv* env);
 
-    Status _close_jni_scanner();
-
     std::map<std::string, std::string> _scanner_params;
     std::vector<JniColumn> _jni_columns;
     Block _jni_block_template;
@@ -114,6 +121,8 @@ class JniTableReader : public TableReader {
     Jni::MethodId _jni_scanner_release_table;
     Jni::MethodId _jni_scanner_get_statistics;
     Jni::MethodId _jni_scanner_set_batch_size;
+    Jni::MethodId _jni_scanner_prepare_for_split;
+    Jni::MethodId _jni_scanner_reset_current_split;
 };
 
 } // namespace doris::format
diff --git a/be/src/format_v2/jni/paimon_jni_reader.cpp b/be/src/format_v2/jni/paimon_jni_reader.cpp
index ca1e602cdfedb9..0cf6360cec6892 100644
--- a/be/src/format_v2/jni/paimon_jni_reader.cpp
+++ b/be/src/format_v2/jni/paimon_jni_reader.cpp
@@ -91,7 +91,6 @@ Status PaimonJniReader::build_scanner_params(std::map<std::string, std::string>*
     const auto& paimon_params = _current_range.table_format_params.paimon_params;
     const auto* paimon_predicate = get_paimon_predicate(_scan_params, paimon_params);
     DORIS_CHECK(paimon_predicate != nullptr);
-    (*params)["paimon_split"] = paimon_params.paimon_split;
     (*params)["paimon_predicate"] = *paimon_predicate;
     (*params)["serialized_table"] = _scan_params->serialized_table;
 
@@ -125,6 +124,57 @@ Status PaimonJniReader::build_scanner_params(std::map<std::string, std::string>*
     return Status::OK();
 }
 
+Status PaimonJniReader::open_jni_scanner_for_split() {
+    if (!jni_scanner_opened()) {
+        RETURN_IF_ERROR(JniTableReader::open_jni_scanner_for_split());
+    }
+    return _prepare_for_split();
+}
+
+Status PaimonJniReader::close_jni_scanner_for_split() {
+    RETURN_IF_ERROR(_reset_current_split());
+    reset_jni_eof();
+    return Status::OK();
+}
+
+Status PaimonJniReader::_prepare_for_split() {
+    DORIS_CHECK(jni_scanner_opened());
+    DORIS_CHECK(!_current_split_prepared);
+    std::map<std::string, std::string> split_params;
+    split_params["paimon_split"] = _current_range.table_format_params.paimon_params.paimon_split;
+
+    JNIEnv* env = nullptr;
+    RETURN_IF_ERROR(Jni::Env::Get(&env));
+    Jni::LocalObject hashmap_object;
+    RETURN_IF_ERROR(Jni::Util::convert_to_java_map(env, split_params, &hashmap_object));
+    RETURN_IF_ERROR(jni_scanner_obj()
+                            .call_void_method(env, jni_scanner_prepare_for_split())
+                            .with_arg(hashmap_object)
+                            .call());
+    RETURN_ERROR_IF_EXC(env);
+    _current_split_prepared = true;
+    reset_jni_eof();
+    return Status::OK();
+}
+
+Status PaimonJniReader::close() {
+    RETURN_IF_ERROR(_reset_current_split());
+    return JniTableReader::close();
+}
+
+Status PaimonJniReader::_reset_current_split() {
+    if (!_current_split_prepared) {
+        return Status::OK();
+    }
+    JNIEnv* env = nullptr;
+    RETURN_IF_ERROR(Jni::Env::Get(&env));
+    RETURN_IF_ERROR(
+            jni_scanner_obj().call_void_method(env, jni_scanner_reset_current_split()).call());
+    RETURN_ERROR_IF_EXC(env);
+    _current_split_prepared = false;
+    return Status::OK();
+}
+
 std::string PaimonJniReader::build_default_io_manager_tmp_dirs(
         const std::vector<StorePath>& store_paths) {
     std::vector<std::string> tmp_dirs;
diff --git a/be/src/format_v2/jni/paimon_jni_reader.h b/be/src/format_v2/jni/paimon_jni_reader.h
index 307880815654fd..ddd487747cd973 100644
--- a/be/src/format_v2/jni/paimon_jni_reader.h
+++ b/be/src/format_v2/jni/paimon_jni_reader.h
@@ -32,6 +32,7 @@ namespace doris::format::paimon {
 class PaimonJniReader final : public format::JniTableReader {
 public:
     ~PaimonJniReader() override = default;
+    Status close() override;
 
 #ifdef BE_TEST
     void TEST_set_scan_params(TFileScanRangeParams* params) { _scan_params = params; }
@@ -50,9 +51,15 @@ 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;
+    Status open_jni_scanner_for_split() override;
+    Status close_jni_scanner_for_split() override;
 
 private:
+    Status _prepare_for_split();
+    Status _reset_current_split();
     static std::string build_default_io_manager_tmp_dirs(const std::vector<StorePath>& store_paths);
+
+    bool _current_split_prepared = false;
 };
 
 } // namespace doris::format::paimon
diff --git a/fe/be-java-extensions/java-common/src/main/java/org/apache/doris/common/jni/JniScanner.java b/fe/be-java-extensions/java-common/src/main/java/org/apache/doris/common/jni/JniScanner.java
index 4e2e23b5ae0851..60dd1d7f5b23d8 100644
--- a/fe/be-java-extensions/java-common/src/main/java/org/apache/doris/common/jni/JniScanner.java
+++ b/fe/be-java-extensions/java-common/src/main/java/org/apache/doris/common/jni/JniScanner.java
@@ -52,6 +52,14 @@ public abstract class JniScanner {
     // Close JniScanner and release resources
     public abstract void close() throws IOException;
 
+    public void prepareForSplit(Map<String, String> splitParams) throws IOException {
+        throw new UnsupportedOperationException(getClass().getName() + " does not support split preparation");
+    }
+
+    public void resetCurrentSplit() throws IOException {
+        throw new UnsupportedOperationException(getClass().getName() + " does not support split reset");
+    }
+
     // Scan data and save as vector table
     protected abstract int getNext() throws IOException;
 
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 cd31a0815e8ae4..2fee6b8de12125 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
@@ -50,7 +50,6 @@
 import java.nio.file.Files;
 import java.nio.file.Paths;
 import java.util.Arrays;
-import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Locale;
@@ -70,11 +69,13 @@ public class PaimonJniScanner extends JniScanner {
     static final String JNI_IO_MANAGER_TMP_DIR = "paimon.doris.jni_io_manager.tmp_dir";
     static final String JNI_IO_MANAGER_IMPL_CLASS = "paimon.doris.jni_io_manager.impl_class";
     private static final AtomicInteger ACTIVE_SCANNERS = new AtomicInteger();
+    static final String DORIS_ENABLE_FILE_READER_ASYNC = "paimon.jni.enable_file_reader_async";
+    static final String MAX_ASYNC_READ_THRESHOLD = Long.MAX_VALUE + "b"; // max threshold means disable
 
     private final Map<String, String> params;
     private final Map<String, String> hadoopOptionParams;
-    private final String paimonSplit;
     private final String paimonPredicate;
+    private String currentPaimonSplit;
     private Table table;
     private RecordReader<InternalRow> reader;
     private IOManager ioManager;
@@ -82,6 +83,7 @@ public class PaimonJniScanner extends JniScanner {
     private final PaimonColumnValue columnValue = new PaimonColumnValue();
     private List<String> paimonAllFieldNames;
     private List<DataType> paimonDataTypeList;
+    private int[] projected;
     private RecordReader.RecordIterator<InternalRow> recordIterator = null;
     private final ClassLoader classLoader;
     private PreExecutionAuthenticator preExecutionAuthenticator;
@@ -107,7 +109,6 @@ public PaimonJniScanner(int batchSize, Map<String, String> params) {
         for (int i = 0; i < requiredTypes.length; i++) {
             columnTypes[i] = ColumnType.parseType(requiredFields[i], requiredTypes[i]);
         }
-        paimonSplit = params.get("paimon_split");
         paimonPredicate = params.get("paimon_predicate");
         String timeZone = params.getOrDefault("time_zone", TimeZone.getDefault().getID());
         columnValue.setTimeZone(timeZone);
@@ -132,7 +133,8 @@ public void open() throws IOException {
             preExecutionAuthenticator.execute(() -> {
                 PaimonJdbcDriverUtils.registerDriverIfNeeded(params, classLoader);
                 initTable();
-                initReader();
+                initProjection();
+                initIOManagerIfNeeded();
                 return null;
             });
             resetDatetimeV2Precision();
@@ -150,8 +152,54 @@ public void open() throws IOException {
         }
     }
 
-    private void initReader() throws IOException {
-        ReadBuilder readBuilder = table.newReadBuilder();
+    @Override
+    public void prepareForSplit(Map<String, String> splitParams) throws IOException {
+        Preconditions.checkArgument(splitParams != null, "Paimon split params are required");
+        String paimonSplit = splitParams.get("paimon_split");
+        Preconditions.checkArgument(paimonSplit != null && !paimonSplit.isEmpty(), "paimon_split is required");
+        try {
+            Thread.currentThread().setContextClassLoader(classLoader);
+            preExecutionAuthenticator.execute(() -> {
+                resetCurrentSplit();
+                currentPaimonSplit = paimonSplit;
+                initReader();
+                return null;
+            });
+        } catch (IOException e) {
+            throw e;
+        } catch (Exception e) {
+            throw new IOException(e);
+        }
+    }
+
+    @Override
+    public void resetCurrentSplit() throws IOException {
+        IOException exception = null;
+        try {
+            releaseRecordIterator();
+        } catch (RuntimeException e) {
+            exception = new IOException("Failed to release Paimon record iterator", e);
+        }
+        if (reader != null) {
+            try {
+                reader.close();
+            } catch (IOException e) {
+                if (exception == null) {
+                    exception = e;
+                } else {
+                    exception.addSuppressed(e);
+                }
+            } finally {
+                reader = null;
+            }
+        }
+        currentPaimonSplit = null;
+        if (exception != null) {
+            throw exception;
+        }
+    }
+
+    private void initProjection() throws IOException {
         if (this.fields.length > this.paimonAllFieldNames.size()) {
             throw new IOException(
                     String.format(
@@ -159,12 +207,19 @@ private void initReader() throws IOException {
                                     + " Please refresh table and try again",
                             fields.length, paimonAllFieldNames.size()));
         }
-        int[] projected = getProjected();
+        projected = getProjected();
+        paimonDataTypeList =
+                Arrays.stream(projected).mapToObj(i -> table.rowType().getTypeAt(i)).collect(Collectors.toList());
+    }
+
+    private void initReader() throws IOException {
+        Preconditions.checkState(table != null, "Paimon scanner is not opened");
+        Preconditions.checkState(projected != null, "Paimon scanner projection is not initialized");
+        Preconditions.checkState(currentPaimonSplit != null, "Paimon split is not prepared");
+        ReadBuilder readBuilder = table.newReadBuilder();
         readBuilder.withProjection(projected);
         readBuilder.withFilter(getPredicates());
         reader = newReadWithOptionalIOManager(readBuilder).executeFilter().createReader(getSplit());
-        paimonDataTypeList =
-                Arrays.stream(projected).mapToObj(i -> table.rowType().getTypeAt(i)).collect(Collectors.toList());
     }
 
     private TableRead newReadWithOptionalIOManager(ReadBuilder readBuilder) throws IOException {
@@ -172,11 +227,18 @@ private TableRead newReadWithOptionalIOManager(ReadBuilder readBuilder) throws I
         if (!isIOManagerEnabled(params)) {
             return tableRead;
         }
+        Preconditions.checkState(ioManager != null, "Paimon JNI IOManager is not initialized");
+        return tableRead.withIOManager(ioManager);
+    }
+
+    private void initIOManagerIfNeeded() throws IOException {
+        if (!isIOManagerEnabled(params) || ioManager != null) {
+            return;
+        }
         ioManagerTempDirs = getIOManagerTempDirs(params);
         ioManager = createIOManager(ioManagerTempDirs, getIOManagerImplClass(params));
         LOG.info("Enable Paimon JNI IOManager with temp dirs: {}, implementation: {}",
                 ioManagerTempDirs, ioManager.getClass().getName());
-        return tableRead.withIOManager(ioManager);
     }
 
     static boolean isIOManagerEnabled(Map<String, String> params) {
@@ -278,7 +340,8 @@ private List<Predicate> getPredicates() {
     }
 
     private Split getSplit() {
-        Split split = PaimonUtils.deserialize(paimonSplit);
+        Preconditions.checkState(currentPaimonSplit != null, "Paimon split is not prepared");
+        Split split = PaimonUtils.deserialize(currentPaimonSplit);
         if (LOG.isDebugEnabled()) {
             LOG.debug("split:{}", split);
         }
@@ -305,15 +368,11 @@ private void resetDatetimeV2Precision() {
     public void close() throws IOException {
         IOException exception = null;
         try {
-            if (reader != null) {
-                try {
-                    reader.close();
-                } catch (IOException e) {
-                    exception = e;
-                } finally {
-                    reader = null;
-                }
-            }
+            resetCurrentSplit();
+        } catch (IOException e) {
+            exception = e;
+        }
+        try {
             if (ioManager != null) {
                 try {
                     ioManager.close();
@@ -336,9 +395,20 @@ public void close() throws IOException {
         }
     }
 
+    private void releaseRecordIterator() {
+        if (recordIterator != null) {
+            try {
+                recordIterator.releaseBatch();
+            } finally {
+                recordIterator = null;
+            }
+        }
+    }
+
     private int readAndProcessNextBatch() throws IOException {
         int rows = 0;
         try {
+            Preconditions.checkState(reader != null, "Paimon split is not prepared");
             if (recordIterator == null) {
                 recordIterator = readBatchWithMetrics();
             }
@@ -365,7 +435,7 @@ private int readAndProcessNextBatch() throws IOException {
                 }
                 appendDataTime += System.nanoTime() - startTime;
 
-                recordIterator.releaseBatch();
+                releaseRecordIterator();
                 recordIterator = readBatchWithMetrics();
             }
             if (fields.length == 0 && rows > 0) {
@@ -373,10 +443,14 @@ private int readAndProcessNextBatch() throws IOException {
             }
             rowsRead += rows;
         } catch (Exception e) {
-            close();
             LOG.warn("Failed to get the next batch of paimon. "
                             + "split: {}, requiredFieldNames: {}, paimonAllFieldNames: {}, dataType: {}",
-                    getSplit(), params.get("required_fields"), paimonAllFieldNames, paimonDataTypeList, e);
+                    currentPaimonSplit, params.get("required_fields"), paimonAllFieldNames, paimonDataTypeList, e);
+            try {
+                close();
+            } catch (IOException closeException) {
+                e.addSuppressed(closeException);
+            }
             throw new IOException(e);
         }
         return rows;
@@ -419,7 +493,7 @@ public Map<String, String> getStatistics() {
         statistics.put("gauge:PaimonJniAsyncReaderThreadCount",
                 String.valueOf(currentAsyncReaderThreadCount()));
         statistics.put("gauge:PaimonJniRequiredFieldCount", String.valueOf(fields.length));
-        statistics.put("counter:PaimonJniSplitEncodedLength", String.valueOf(lengthOfParam("paimon_split")));
+        statistics.put("counter:PaimonJniSplitEncodedLength", String.valueOf(lengthOfCurrentPaimonSplit()));
         statistics.put("counter:PaimonJniPredicateEncodedLength", String.valueOf(lengthOfParam("paimon_predicate")));
         statistics.put("gauge:PaimonJniAsyncThresholdConfigured",
                 hasPaimonOption(FILE_READER_ASYNC_THRESHOLD) ? "1" : "0");
@@ -439,6 +513,10 @@ private int lengthOfParam(String key) {
         return value == null ? 0 : value.length();
     }
 
+    private int lengthOfCurrentPaimonSplit() {
+        return currentPaimonSplit == null ? 0 : currentPaimonSplit.length();
+    }
+
     private boolean hasPaimonOption(String key) {
         return paimonOption(key) != null;
     }
@@ -556,8 +634,7 @@ static Optional<Long> parseDataSizeBytes(String value) {
     private void initTable() {
         Preconditions.checkState(params.containsKey("serialized_table"));
         table = PaimonUtils.deserialize(params.get("serialized_table"));
-        table = table.copy(Collections.singletonMap(
-                CoreOptions.READ_BATCH_SIZE.key(), String.valueOf(batchSize)));
+        table = table.copy(buildTableOptions(table.options()));
         paimonAllFieldNames = PaimonUtils.getFieldNames(this.table.rowType());
         if (LOG.isDebugEnabled()) {
             LOG.debug("paimonAllFieldNames:{}", paimonAllFieldNames);
@@ -571,4 +648,12 @@ private static String[] splitParam(String value, String delimiter) {
         return value.split(delimiter);
     }
 
+    private Map<String, String> buildTableOptions(Map<String, String> tableOptions) {
+        Map<String, String> options = new HashMap<>(tableOptions);
+        options.put(CoreOptions.READ_BATCH_SIZE.key(), String.valueOf(batchSize));
+        if (Boolean.parseBoolean(params.getOrDefault(DORIS_ENABLE_FILE_READER_ASYNC, "true")) == false) {
+            options.put(CoreOptions.FILE_READER_ASYNC_THRESHOLD.key(), MAX_ASYNC_READ_THRESHOLD);
+        }
+        return options;
+    }
 }
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 c6c90389fe8af0..9e8e22a445d35d 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
@@ -17,11 +17,13 @@
 
 package org.apache.doris.paimon;
 
+import org.apache.paimon.data.InternalRow;
 import org.apache.paimon.disk.BufferFileReader;
 import org.apache.paimon.disk.BufferFileWriter;
 import org.apache.paimon.disk.FileIOChannel;
 import org.apache.paimon.disk.IOManager;
 import org.apache.paimon.disk.IOManagerImpl;
+import org.apache.paimon.reader.RecordReader;
 import org.apache.paimon.table.Table;
 import org.junit.Assert;
 import org.junit.Rule;
@@ -37,6 +39,7 @@
 import java.util.Map;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 public class PaimonJniScannerTest {
     @Rule
@@ -102,10 +105,10 @@ public void testCloseCleansIOManagerTempDirectory() throws Exception {
     @Test
     public void testStatisticsIncludePaimonDiagnostics() throws Exception {
         Map<String, String> params = createBaseParams();
-        params.put("paimon_split", "encoded-split");
         params.put("paimon_predicate", "encoded-predicate");
         PaimonJniScanner scanner = new PaimonJniScanner(128, params);
         setTableOptions(scanner, Collections.singletonMap("file-reader-async-threshold", "10 MiB"));
+        setCurrentPaimonSplit(scanner, "encoded-split");
 
         Map<String, String> statistics = scanner.getStatistics();
 
@@ -160,11 +163,54 @@ public void testParseDataSizeBytes() {
         Assert.assertFalse(PaimonJniScanner.parseDataSizeBytes("unknown").isPresent());
     }
 
+    @Test
+    public void testCloseReleasesActiveRecordIterator() throws Exception {
+        PaimonJniScanner scanner = new PaimonJniScanner(128, createBaseParams());
+        AtomicBoolean released = new AtomicBoolean(false);
+        RecordReader.RecordIterator<InternalRow> recordIterator = createRecordIterator(released);
+
+        Field recordIteratorField = PaimonJniScanner.class.getDeclaredField("recordIterator");
+        recordIteratorField.setAccessible(true);
+        recordIteratorField.set(scanner, recordIterator);
+
+        scanner.close();
+
+        Assert.assertTrue(released.get());
+        Assert.assertNull(recordIteratorField.get(scanner));
+    }
+
+    @Test
+    public void testResetCurrentSplitReleasesActiveRecordIterator() throws Exception {
+        PaimonJniScanner scanner = new PaimonJniScanner(128, createBaseParams());
+        AtomicBoolean released = new AtomicBoolean(false);
+        RecordReader.RecordIterator<InternalRow> recordIterator = createRecordIterator(released);
+
+        Field recordIteratorField = PaimonJniScanner.class.getDeclaredField("recordIterator");
+        recordIteratorField.setAccessible(true);
+        recordIteratorField.set(scanner, recordIterator);
+
+        scanner.resetCurrentSplit();
+
+        Assert.assertTrue(released.get());
+        Assert.assertNull(recordIteratorField.get(scanner));
+    }
+
+    @Test
+    public void testPrepareForSplitRequiresSplitParam() throws Exception {
+        PaimonJniScanner scanner = new PaimonJniScanner(128, createBaseParams());
+
+        try {
+            scanner.prepareForSplit(new HashMap<>());
+            Assert.fail("Expected prepareForSplit to reject missing paimon_split");
+        } catch (IllegalArgumentException e) {
+            Assert.assertTrue(e.getMessage().contains("paimon_split"));
+        }
+    }
+
     private Map<String, String> createBaseParams() {
         Map<String, String> params = new HashMap<>();
         params.put("required_fields", "");
         params.put("columns_types", "");
-        params.put("paimon_split", "");
         params.put("paimon_predicate", "");
         return params;
     }
@@ -185,6 +231,26 @@ private void setTableOptions(PaimonJniScanner scanner, Map<String, String> optio
         tableField.set(scanner, table);
     }
 
+    private void setCurrentPaimonSplit(PaimonJniScanner scanner, String split) throws Exception {
+        Field currentPaimonSplitField = PaimonJniScanner.class.getDeclaredField("currentPaimonSplit");
+        currentPaimonSplitField.setAccessible(true);
+        currentPaimonSplitField.set(scanner, split);
+    }
+
+    private RecordReader.RecordIterator<InternalRow> createRecordIterator(AtomicBoolean released) {
+        return new RecordReader.RecordIterator<InternalRow>() {
+            @Override
+            public InternalRow next() {
+                return null;
+            }
+
+            @Override
+            public void releaseBatch() {
+                released.set(true);
+            }
+        };
+    }
+
     public static class TestIOManager implements IOManager {
         private final String[] tempDirs;
 
diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonScanNode.java
index f8549855ea6e50..da3e50a6be46e4 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonScanNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonScanNode.java
@@ -93,13 +93,15 @@ public class PaimonScanNode extends FileQueryScanNode {
     private static final String DORIS_END_TIMESTAMP = "endTimestamp";
     private static final String DORIS_INCREMENTAL_BETWEEN_SCAN_MODE = "incrementalBetweenScanMode";
     private static final String PAIMON_PROPERTY_PREFIX = "paimon.";
+    private static final String DORIS_ENABLE_FILE_READER_ASYNC = "jni.enable_file_reader_async";
     private static final String DORIS_ENABLE_JNI_IO_MANAGER = "doris.enable_jni_io_manager";
     private static final String DORIS_JNI_IO_MANAGER_TMP_DIR = "doris.jni_io_manager.tmp_dir";
     private static final String DORIS_JNI_IO_MANAGER_IMPL_CLASS = "doris.jni_io_manager.impl_class";
     private static final List<String> BACKEND_PAIMON_OPTIONS = Arrays.asList(
             DORIS_ENABLE_JNI_IO_MANAGER,
             DORIS_JNI_IO_MANAGER_TMP_DIR,
-            DORIS_JNI_IO_MANAGER_IMPL_CLASS);
+            DORIS_JNI_IO_MANAGER_IMPL_CLASS,
+            DORIS_ENABLE_FILE_READER_ASYNC);
     private static final String PAIMON_BINLOG_SYSTEM_TABLE_TYPE = "binlog";
     private static final String PAIMON_AUDIT_LOG_SYSTEM_TABLE_TYPE = "audit_log";