This is an automated email from the ASF dual-hosted git repository.
zhangstar333 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new f81780bbfe9 [opt](paimon) control paimon jni reader async file reader
(#65365)
f81780bbfe9 is described below
commit f81780bbfe99f99bfa387c2bc3d58ceed7f6bef0
Author: zhangstar333 <[email protected]>
AuthorDate: Fri Jul 10 17:02:43 2026 +0800
[opt](paimon) control paimon jni reader async file reader (#65365)
### What problem does this PR solve?
Problem Summary:
add a property for paimon jni reader to control when read orc data
files, whether is need use async threads.
the default value is true keep with paimon options.
if user set it to false like below, we will pass the Long.MAX_VALUE as
threshold to paimon table options, those would means disable.
2. add releaseRecordIterator() in close function, so if cancel query
this iterator also will be release not cause leak.
```
CREATE CATALOG paimon_catalog PROPERTIES (
"type" = "paimon",
xxxxxx ......
"paimon.jni.enable_file_reader_async" = "false"
);
```
---
.../org/apache/doris/paimon/PaimonJniScanner.java | 37 +++++++++++++++++++---
.../apache/doris/paimon/PaimonJniScannerTest.java | 30 ++++++++++++++++++
.../datasource/paimon/source/PaimonScanNode.java | 4 ++-
3 files changed, 65 insertions(+), 6 deletions(-)
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 cd31a0815e8..0cc3e814082 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.lang.reflect.InvocationTargetException;
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,6 +69,8 @@ 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;
@@ -305,11 +306,20 @@ public class PaimonJniScanner extends JniScanner {
public void close() throws IOException {
IOException exception = null;
try {
+ try {
+ releaseRecordIterator();
+ } catch (RuntimeException e) {
+ exception = new IOException("Failed to release Paimon record
iterator", e);
+ }
if (reader != null) {
try {
reader.close();
} catch (IOException e) {
- exception = e;
+ if (exception == null) {
+ exception = e;
+ } else {
+ exception.addSuppressed(e);
+ }
} finally {
reader = null;
}
@@ -336,6 +346,16 @@ public class PaimonJniScanner extends JniScanner {
}
}
+ private void releaseRecordIterator() {
+ if (recordIterator != null) {
+ try {
+ recordIterator.releaseBatch();
+ } finally {
+ recordIterator = null;
+ }
+ }
+ }
+
private int readAndProcessNextBatch() throws IOException {
int rows = 0;
try {
@@ -365,7 +385,7 @@ public class PaimonJniScanner extends JniScanner {
}
appendDataTime += System.nanoTime() - startTime;
- recordIterator.releaseBatch();
+ releaseRecordIterator();
recordIterator = readBatchWithMetrics();
}
if (fields.length == 0 && rows > 0) {
@@ -556,8 +576,7 @@ public class PaimonJniScanner extends JniScanner {
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 +590,12 @@ public class PaimonJniScanner extends JniScanner {
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 c6c90389fe8..6f893b38b1c 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.HashMap;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
public class PaimonJniScannerTest {
@Rule
@@ -160,6 +163,33 @@ public class PaimonJniScannerTest {
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 =
+ new RecordReader.RecordIterator<InternalRow>() {
+ @Override
+ public InternalRow next() {
+ return null;
+ }
+
+ @Override
+ public void releaseBatch() {
+ released.set(true);
+ }
+ };
+
+ Field recordIteratorField =
PaimonJniScanner.class.getDeclaredField("recordIterator");
+ recordIteratorField.setAccessible(true);
+ recordIteratorField.set(scanner, recordIterator);
+
+ scanner.close();
+
+ Assert.assertTrue(released.get());
+ Assert.assertNull(recordIteratorField.get(scanner));
+ }
+
private Map<String, String> createBaseParams() {
Map<String, String> params = new HashMap<>();
params.put("required_fields", "");
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 f8549855ea6..da3e50a6be4 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";
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
