This is an automated email from the ASF dual-hosted git repository.
Gabriel39 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 74e005f6905 [improvement](paimon) Add Paimon JNI scanner observability
(#65354)
74e005f6905 is described below
commit 74e005f69050d993d1b480d13e55a4072554c4ba
Author: Gabriel <[email protected]>
AuthorDate: Wed Jul 8 16:46:21 2026 +0800
[improvement](paimon) Add Paimon JNI scanner observability (#65354)
Problem Summary: Paimon JNI scans can create Paimon SDK internal async
file readers, especially for primary-key merge reads over large ORC
files. Doris scanner concurrency limits do not directly expose those
internal async reader threads or Java heap pressure in the query
profile. This change adds lightweight profile metrics from the Paimon
JNI scanner for active scanner counts, async reader thread counts, JVM
heap/non-heap usage, split and predicate diagnostic sizes, async
threshold configuration visibility, and readBatch/open timing counters.
---
be/src/format/jni/jni_reader.cpp | 36 +++-
be/src/format_v2/jni/jni_table_reader.cpp | 79 ++++++++
be/src/format_v2/jni/jni_table_reader.h | 2 +
.../org/apache/doris/paimon/PaimonJniScanner.java | 221 +++++++++++++++++++--
.../apache/doris/paimon/PaimonJniScannerTest.java | 84 +++++++-
5 files changed, 396 insertions(+), 26 deletions(-)
diff --git a/be/src/format/jni/jni_reader.cpp b/be/src/format/jni/jni_reader.cpp
index e45cffd4cc3..546d8446d58 100644
--- a/be/src/format/jni/jni_reader.cpp
+++ b/be/src/format/jni/jni_reader.cpp
@@ -424,6 +424,9 @@ void JniReader::_collect_profile_before_close() {
return;
}
+ const auto update_peak = [](int64_t previous, int64_t current) {
+ return current > previous;
+ };
for (const auto& metric : statistics_result) {
std::vector<std::string> type_and_name = split(metric.first, ":");
if (type_and_name.size() != 2) {
@@ -431,22 +434,49 @@ void JniReader::_collect_profile_before_close() {
<< "'metricType:metricName'";
continue;
}
- long metric_value = std::stol(metric.second);
+ int64_t metric_value = std::stoll(metric.second);
RuntimeProfile::Counter* scanner_counter;
if (type_and_name[0] == "timer") {
scanner_counter =
ADD_CHILD_TIMER(_profile, type_and_name[1],
_connector_name.c_str());
+ COUNTER_UPDATE(scanner_counter, metric_value);
} else if (type_and_name[0] == "counter") {
scanner_counter = ADD_CHILD_COUNTER(_profile,
type_and_name[1], TUnit::UNIT,
_connector_name.c_str());
+ COUNTER_UPDATE(scanner_counter, metric_value);
} else if (type_and_name[0] == "bytes") {
scanner_counter = ADD_CHILD_COUNTER(_profile,
type_and_name[1], TUnit::BYTES,
_connector_name.c_str());
+ COUNTER_UPDATE(scanner_counter, metric_value);
+ } else if (type_and_name[0] == "timer_gauge") {
+ scanner_counter =
+ ADD_CHILD_TIMER(_profile, type_and_name[1],
_connector_name.c_str());
+ COUNTER_SET(scanner_counter, metric_value);
+ } else if (type_and_name[0] == "gauge") {
+ scanner_counter = ADD_CHILD_COUNTER(_profile,
type_and_name[1], TUnit::UNIT,
+ _connector_name.c_str());
+ COUNTER_SET(scanner_counter, metric_value);
+ } else if (type_and_name[0] == "bytes_gauge") {
+ scanner_counter = ADD_CHILD_COUNTER(_profile,
type_and_name[1], TUnit::BYTES,
+ _connector_name.c_str());
+ COUNTER_SET(scanner_counter, metric_value);
+ } else if (type_and_name[0] == "timer_peak") {
+ auto* scanner_peak_counter = _profile->add_conditition_counter(
+ type_and_name[1], TUnit::TIME_NS, update_peak,
_connector_name.c_str());
+ scanner_peak_counter->conditional_update(metric_value,
metric_value);
+ } else if (type_and_name[0] == "peak") {
+ auto* scanner_peak_counter = _profile->add_conditition_counter(
+ type_and_name[1], TUnit::UNIT, update_peak,
_connector_name.c_str());
+ scanner_peak_counter->conditional_update(metric_value,
metric_value);
+ } else if (type_and_name[0] == "bytes_peak") {
+ auto* scanner_peak_counter = _profile->add_conditition_counter(
+ type_and_name[1], TUnit::BYTES, update_peak,
_connector_name.c_str());
+ scanner_peak_counter->conditional_update(metric_value,
metric_value);
} else {
- LOG(WARNING) << "Type of JNI Scanner metric should be timer,
counter or bytes";
+ LOG(WARNING) << "Type of JNI Scanner metric should be timer,
counter, bytes, "
+ << "timer_gauge, gauge, bytes_gauge, timer_peak,
peak or bytes_peak";
continue;
}
- COUNTER_UPDATE(scanner_counter, metric_value);
}
}
}
diff --git a/be/src/format_v2/jni/jni_table_reader.cpp
b/be/src/format_v2/jni/jni_table_reader.cpp
index 01c2dd6ecd0..deefe26837e 100644
--- a/be/src/format_v2/jni/jni_table_reader.cpp
+++ b/be/src/format_v2/jni/jni_table_reader.cpp
@@ -20,6 +20,7 @@
#include <utility>
#include "common/cast_set.h"
+#include "common/logging.h"
#include "core/block/block.h"
#include "exprs/vexpr_context.h"
#include "runtime/descriptors.h"
@@ -176,6 +177,83 @@ Status JniTableReader::finalize_jni_block(Block*
jni_block, Block* output_block,
return Status::OK();
}
+Status JniTableReader::_get_statistics(JNIEnv* env, std::map<std::string,
std::string>* result) {
+ DORIS_CHECK(result != nullptr);
+ result->clear();
+ Jni::LocalObject metrics;
+ RETURN_IF_ERROR(
+ _jni_scanner_obj.call_object_method(env,
_jni_scanner_get_statistics).call(&metrics));
+ RETURN_IF_ERROR(Jni::Util::convert_to_cpp_map(env, metrics, result));
+ return Status::OK();
+}
+
+void JniTableReader::_collect_jni_scanner_profile(JNIEnv* env) {
+ if (_scanner_profile == nullptr) {
+ return;
+ }
+
+ std::map<std::string, std::string> statistics_result;
+ Status st = _get_statistics(env, &statistics_result);
+ if (!st) {
+ LOG(WARNING) << "failed to get_statistics when collect profile: " <<
st;
+ return;
+ }
+
+ const auto connector_name = _connector_name();
+ const auto update_peak = [](int64_t previous, int64_t current) { return
current > previous; };
+ for (const auto& metric : statistics_result) {
+ std::vector<std::string> type_and_name = split(metric.first, ":");
+ if (type_and_name.size() != 2) {
+ LOG(WARNING) << "Name of JNI Scanner metric should be pattern like
"
+ << "'metricType:metricName'";
+ continue;
+ }
+ int64_t metric_value = std::stoll(metric.second);
+ RuntimeProfile::Counter* scanner_counter;
+ if (type_and_name[0] == "timer") {
+ scanner_counter =
+ ADD_CHILD_TIMER(_scanner_profile, type_and_name[1],
connector_name.c_str());
+ COUNTER_UPDATE(scanner_counter, metric_value);
+ } else if (type_and_name[0] == "counter") {
+ scanner_counter = ADD_CHILD_COUNTER(_scanner_profile,
type_and_name[1], TUnit::UNIT,
+ connector_name.c_str());
+ COUNTER_UPDATE(scanner_counter, metric_value);
+ } else if (type_and_name[0] == "bytes") {
+ scanner_counter = ADD_CHILD_COUNTER(_scanner_profile,
type_and_name[1], TUnit::BYTES,
+ connector_name.c_str());
+ COUNTER_UPDATE(scanner_counter, metric_value);
+ } else if (type_and_name[0] == "timer_gauge") {
+ scanner_counter =
+ ADD_CHILD_TIMER(_scanner_profile, type_and_name[1],
connector_name.c_str());
+ COUNTER_SET(scanner_counter, metric_value);
+ } else if (type_and_name[0] == "gauge") {
+ scanner_counter = ADD_CHILD_COUNTER(_scanner_profile,
type_and_name[1], TUnit::UNIT,
+ connector_name.c_str());
+ COUNTER_SET(scanner_counter, metric_value);
+ } else if (type_and_name[0] == "bytes_gauge") {
+ scanner_counter = ADD_CHILD_COUNTER(_scanner_profile,
type_and_name[1], TUnit::BYTES,
+ connector_name.c_str());
+ COUNTER_SET(scanner_counter, metric_value);
+ } else if (type_and_name[0] == "timer_peak") {
+ auto* scanner_peak_counter =
_scanner_profile->add_conditition_counter(
+ type_and_name[1], TUnit::TIME_NS, update_peak,
connector_name.c_str());
+ scanner_peak_counter->conditional_update(metric_value,
metric_value);
+ } else if (type_and_name[0] == "peak") {
+ auto* scanner_peak_counter =
_scanner_profile->add_conditition_counter(
+ type_and_name[1], TUnit::UNIT, update_peak,
connector_name.c_str());
+ scanner_peak_counter->conditional_update(metric_value,
metric_value);
+ } else if (type_and_name[0] == "bytes_peak") {
+ auto* scanner_peak_counter =
_scanner_profile->add_conditition_counter(
+ type_and_name[1], TUnit::BYTES, update_peak,
connector_name.c_str());
+ scanner_peak_counter->conditional_update(metric_value,
metric_value);
+ } else {
+ LOG(WARNING) << "Type of JNI Scanner metric should be timer,
counter, bytes, "
+ << "timer_gauge, gauge, bytes_gauge, timer_peak, peak
or bytes_peak";
+ continue;
+ }
+ }
+}
+
Status JniTableReader::build_jni_columns(std::vector<JniColumn>* columns)
const {
DORIS_CHECK(columns != nullptr);
columns->clear();
@@ -240,6 +318,7 @@ Status JniTableReader::_close_jni_scanner() {
_jni_scanner_open_watcher + _fill_block_watcher +
_java_scan_watcher,
self_split_weight());
}
+ _collect_jni_scanner_profile(env);
// _fill_jni_block may fail before releasing the current Java table.
JniScanner::releaseTable()
// is idempotent, so closing the split always releases it.
diff --git a/be/src/format_v2/jni/jni_table_reader.h
b/be/src/format_v2/jni/jni_table_reader.h
index 1317661e588..cfe1eb32c69 100644
--- a/be/src/format_v2/jni/jni_table_reader.h
+++ b/be/src/format_v2/jni/jni_table_reader.h
@@ -78,6 +78,8 @@ private:
// 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();
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 4593439ec4c..f6dd3334e76 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
@@ -40,6 +40,11 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
+import java.lang.management.ManagementFactory;
+import java.lang.management.MemoryMXBean;
+import java.lang.management.MemoryUsage;
+import java.lang.management.ThreadInfo;
+import java.lang.management.ThreadMXBean;
import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;
import java.nio.file.Files;
@@ -48,16 +53,23 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
+import java.util.Locale;
import java.util.Map;
+import java.util.Optional;
import java.util.TimeZone;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
public class PaimonJniScanner extends JniScanner {
private static final Logger LOG =
LoggerFactory.getLogger(PaimonJniScanner.class);
private static final String HADOOP_OPTION_PREFIX = "hadoop.";
+ private static final String PAIMON_OPTION_PREFIX = "paimon.";
+ private static final String ASYNC_READER_THREAD_NAME_PREFIX =
"paimon-reader-async-thread";
+ private static final String FILE_READER_ASYNC_THRESHOLD =
"file-reader-async-threshold";
static final String ENABLE_JNI_IO_MANAGER =
"paimon.doris.enable_jni_io_manager";
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();
private final Map<String, String> params;
private final Map<String, String> hadoopOptionParams;
@@ -73,6 +85,12 @@ public class PaimonJniScanner extends JniScanner {
private RecordReader.RecordIterator<InternalRow> recordIterator = null;
private final ClassLoader classLoader;
private PreExecutionAuthenticator preExecutionAuthenticator;
+ private boolean scannerCounted;
+ private long openTimeNanos;
+ private long readBatchTimeNanos;
+ private long readBatchCalls;
+ private long emptyReadBatchCalls;
+ private long rowsRead;
public PaimonJniScanner(int batchSize, Map<String, String> params) {
this.classLoader = this.getClass().getClassLoader();
@@ -103,6 +121,8 @@ public class PaimonJniScanner extends JniScanner {
@Override
public void open() throws IOException {
+ markScannerOpenedForMetrics();
+ long startTime = System.nanoTime();
try {
// When the user does not specify hive-site.xml, Paimon will look
for the file from the classpath:
// org.apache.paimon.hive.HiveCatalog.createHiveConf:
@@ -125,6 +145,8 @@ public class PaimonJniScanner extends JniScanner {
}
LOG.warn("Failed to open paimon_scanner: " + e.getMessage(), e);
throw new RuntimeException(e);
+ } finally {
+ openTimeNanos += System.nanoTime() - startTime;
}
}
@@ -273,28 +295,32 @@ public class PaimonJniScanner extends JniScanner {
@Override
public void close() throws IOException {
IOException exception = null;
- if (reader != null) {
- try {
- reader.close();
- } catch (IOException e) {
- exception = e;
- } finally {
- reader = null;
+ try {
+ if (reader != null) {
+ try {
+ reader.close();
+ } catch (IOException e) {
+ exception = e;
+ } finally {
+ reader = null;
+ }
}
- }
- if (ioManager != null) {
- try {
- ioManager.close();
- } catch (Exception e) {
- LOG.warn("Failed to close Paimon JNI IOManager, temp dirs:
{}", ioManagerTempDirs, e);
- if (exception == null) {
- exception = new IOException(e);
- } else {
- exception.addSuppressed(e);
+ if (ioManager != null) {
+ try {
+ ioManager.close();
+ } catch (Exception e) {
+ LOG.warn("Failed to close Paimon JNI IOManager, temp dirs:
{}", ioManagerTempDirs, e);
+ if (exception == null) {
+ exception = new IOException(e);
+ } else {
+ exception.addSuppressed(e);
+ }
+ } finally {
+ ioManager = null;
}
- } finally {
- ioManager = null;
}
+ } finally {
+ markScannerClosedForMetrics();
}
if (exception != null) {
throw exception;
@@ -305,7 +331,7 @@ public class PaimonJniScanner extends JniScanner {
int rows = 0;
try {
if (recordIterator == null) {
- recordIterator = reader.readBatch();
+ recordIterator = readBatchWithMetrics();
}
while (recordIterator != null) {
@@ -324,17 +350,19 @@ public class PaimonJniScanner extends JniScanner {
vectorTable.appendVirtualData(rows);
}
appendDataTime += System.nanoTime() - startTime;
+ rowsRead += rows;
return rows;
}
}
appendDataTime += System.nanoTime() - startTime;
recordIterator.releaseBatch();
- recordIterator = reader.readBatch();
+ recordIterator = readBatchWithMetrics();
}
if (fields.length == 0 && rows > 0) {
vectorTable.appendVirtualData(rows);
}
+ rowsRead += rows;
} catch (Exception e) {
close();
LOG.warn("Failed to get the next batch of paimon. "
@@ -345,6 +373,20 @@ public class PaimonJniScanner extends JniScanner {
return rows;
}
+ private RecordReader.RecordIterator<InternalRow> readBatchWithMetrics()
throws IOException {
+ long startTime = System.nanoTime();
+ try {
+ RecordReader.RecordIterator<InternalRow> iterator =
reader.readBatch();
+ if (iterator == null) {
+ emptyReadBatchCalls++;
+ }
+ return iterator;
+ } finally {
+ readBatchCalls++;
+ readBatchTimeNanos += System.nanoTime() - startTime;
+ }
+ }
+
@Override
protected int getNext() {
try {
@@ -363,10 +405,145 @@ public class PaimonJniScanner extends JniScanner {
@Override
public Map<String, String> getStatistics() {
Map<String, String> statistics = new HashMap<>();
- statistics.put("counter:PaimonJniIOManagerEnabled", ioManager != null
? "1" : "0");
+ statistics.put("gauge:PaimonJniIOManagerEnabled", ioManager != null ?
"1" : "0");
+ statistics.put("gauge:PaimonJniActiveScannerCount",
String.valueOf(ACTIVE_SCANNERS.get()));
+ 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:PaimonJniPredicateEncodedLength",
String.valueOf(lengthOfParam("paimon_predicate")));
+ statistics.put("gauge:PaimonJniAsyncThresholdConfigured",
+ hasPaimonOption(FILE_READER_ASYNC_THRESHOLD) ? "1" : "0");
+
parseDataSizeBytes(paimonOption(FILE_READER_ASYNC_THRESHOLD)).ifPresent(
+ bytes ->
statistics.put("bytes_gauge:PaimonJniAsyncThresholdBytes",
String.valueOf(bytes)));
+ statistics.put("counter:PaimonJniReadBatchCalls",
String.valueOf(readBatchCalls));
+ statistics.put("counter:PaimonJniEmptyReadBatchCalls",
String.valueOf(emptyReadBatchCalls));
+ statistics.put("counter:PaimonJniRowsRead", String.valueOf(rowsRead));
+ statistics.put("timer:PaimonJniScannerOpenTime",
String.valueOf(openTimeNanos));
+ statistics.put("timer:PaimonJniReadBatchTime",
String.valueOf(readBatchTimeNanos));
+ putMemoryStatistics(statistics);
return statistics;
}
+ private int lengthOfParam(String key) {
+ String value = params.get(key);
+ return value == null ? 0 : value.length();
+ }
+
+ private boolean hasPaimonOption(String key) {
+ return paimonOption(key) != null;
+ }
+
+ private String paimonOption(String key) {
+ if (table != null) {
+ String tableOption = table.options().get(key);
+ if (tableOption != null) {
+ return tableOption;
+ }
+ }
+ return params.get(PAIMON_OPTION_PREFIX + key);
+ }
+
+ private static void putMemoryStatistics(Map<String, String> statistics) {
+ MemoryMXBean memoryMXBean = ManagementFactory.getMemoryMXBean();
+ MemoryUsage heapUsage = memoryMXBean.getHeapMemoryUsage();
+ MemoryUsage nonHeapUsage = memoryMXBean.getNonHeapMemoryUsage();
+ statistics.put("bytes_gauge:PaimonJniJvmHeapUsed",
String.valueOf(nonNegative(heapUsage.getUsed())));
+ statistics.put("bytes_gauge:PaimonJniJvmHeapCommitted",
String.valueOf(nonNegative(heapUsage.getCommitted())));
+ statistics.put("bytes_gauge:PaimonJniJvmHeapMax",
String.valueOf(nonNegative(heapUsage.getMax())));
+ statistics.put("bytes_gauge:PaimonJniJvmNonHeapUsed",
String.valueOf(nonNegative(nonHeapUsage.getUsed())));
+ statistics.put("bytes_gauge:PaimonJniJvmNonHeapCommitted",
+ String.valueOf(nonNegative(nonHeapUsage.getCommitted())));
+ statistics.put("bytes_gauge:PaimonJniJvmNonHeapMax",
String.valueOf(nonNegative(nonHeapUsage.getMax())));
+ }
+
+ private static long nonNegative(long value) {
+ return Math.max(value, 0L);
+ }
+
+ private static int currentAsyncReaderThreadCount() {
+ return countThreadsByNamePrefix(ASYNC_READER_THREAD_NAME_PREFIX);
+ }
+
+ static int countThreadsByNamePrefix(String threadNamePrefix) {
+ int count = 0;
+ ThreadMXBean threadMXBean = ManagementFactory.getThreadMXBean();
+ ThreadInfo[] threadInfos =
threadMXBean.getThreadInfo(threadMXBean.getAllThreadIds(), 0);
+ for (ThreadInfo threadInfo : threadInfos) {
+ if (threadInfo != null &&
threadInfo.getThreadName().startsWith(threadNamePrefix)) {
+ count++;
+ }
+ }
+ return count;
+ }
+
+ private void markScannerOpenedForMetrics() {
+ if (!scannerCounted) {
+ scannerCounted = true;
+ ACTIVE_SCANNERS.incrementAndGet();
+ }
+ }
+
+ private void markScannerClosedForMetrics() {
+ if (scannerCounted) {
+ scannerCounted = false;
+ ACTIVE_SCANNERS.decrementAndGet();
+ }
+ }
+
+ static Optional<Long> parseDataSizeBytes(String value) {
+ if (value == null || value.trim().isEmpty()) {
+ return Optional.empty();
+ }
+ String normalized = value.trim().toLowerCase(Locale.ROOT).replace("_",
"").replace(" ", "");
+ int unitStart = 0;
+ while (unitStart < normalized.length()
+ && (Character.isDigit(normalized.charAt(unitStart)) ||
normalized.charAt(unitStart) == '.')) {
+ unitStart++;
+ }
+ if (unitStart == 0) {
+ return Optional.empty();
+ }
+ try {
+ double number = Double.parseDouble(normalized.substring(0,
unitStart));
+ String unit = normalized.substring(unitStart);
+ long multiplier;
+ switch (unit) {
+ case "":
+ case "b":
+ case "byte":
+ case "bytes":
+ multiplier = 1L;
+ break;
+ case "k":
+ case "kb":
+ case "kib":
+ multiplier = 1024L;
+ break;
+ case "m":
+ case "mb":
+ case "mib":
+ multiplier = 1024L * 1024L;
+ break;
+ case "g":
+ case "gb":
+ case "gib":
+ multiplier = 1024L * 1024L * 1024L;
+ break;
+ case "t":
+ case "tb":
+ case "tib":
+ multiplier = 1024L * 1024L * 1024L * 1024L;
+ break;
+ default:
+ return Optional.empty();
+ }
+ return Optional.of((long) (number * multiplier));
+ } catch (NumberFormatException e) {
+ return Optional.empty();
+ }
+ }
+
private void initTable() {
Preconditions.checkState(params.containsKey("serialized_table"));
table = PaimonUtils.deserialize(params.get("serialized_table"));
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 1355d5f9844..4344b89ec40 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
@@ -22,6 +22,7 @@ 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.table.Table;
import org.junit.Assert;
import org.junit.Rule;
import org.junit.Test;
@@ -29,8 +30,12 @@ import org.junit.rules.TemporaryFolder;
import java.io.File;
import java.lang.reflect.Field;
+import java.lang.reflect.Proxy;
+import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
public class PaimonJniScannerTest {
@Rule
@@ -87,12 +92,73 @@ public class PaimonJniScannerTest {
Field ioManagerField =
PaimonJniScanner.class.getDeclaredField("ioManager");
ioManagerField.setAccessible(true);
ioManagerField.set(scanner, ioManager);
- Assert.assertEquals("1",
scanner.getStatistics().get("counter:PaimonJniIOManagerEnabled"));
+ Assert.assertEquals("1",
scanner.getStatistics().get("gauge:PaimonJniIOManagerEnabled"));
scanner.close();
Assert.assertFalse(spillDir.exists());
}
+ @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"));
+
+ Map<String, String> statistics = scanner.getStatistics();
+
+ Assert.assertEquals("0",
statistics.get("gauge:PaimonJniIOManagerEnabled"));
+ Assert.assertEquals("0",
statistics.get("gauge:PaimonJniRequiredFieldCount"));
+ Assert.assertEquals("13",
statistics.get("counter:PaimonJniSplitEncodedLength"));
+ Assert.assertEquals("17",
statistics.get("counter:PaimonJniPredicateEncodedLength"));
+ Assert.assertEquals("1",
statistics.get("gauge:PaimonJniAsyncThresholdConfigured"));
+ Assert.assertEquals(String.valueOf(10L * 1024L * 1024L),
+ statistics.get("bytes_gauge:PaimonJniAsyncThresholdBytes"));
+
Assert.assertTrue(statistics.containsKey("gauge:PaimonJniAsyncReaderThreadCount"));
+
Assert.assertTrue(statistics.containsKey("gauge:PaimonJniActiveScannerCount"));
+
Assert.assertFalse(statistics.containsKey("peak:PaimonJniActiveScannerPeakCount"));
+
Assert.assertFalse(statistics.containsKey("peak:PaimonJniAsyncReaderThreadPeakCount"));
+
Assert.assertTrue(statistics.containsKey("counter:PaimonJniReadBatchCalls"));
+
Assert.assertTrue(statistics.containsKey("timer:PaimonJniScannerOpenTime"));
+
Assert.assertTrue(statistics.containsKey("timer:PaimonJniReadBatchTime"));
+
Assert.assertTrue(Long.parseLong(statistics.get("bytes_gauge:PaimonJniJvmHeapUsed"))
> 0);
+
Assert.assertTrue(Long.parseLong(statistics.get("bytes_gauge:PaimonJniJvmHeapCommitted"))
> 0);
+ }
+
+ @Test
+ public void testCountThreadsByNamePrefix() throws Exception {
+ CountDownLatch started = new CountDownLatch(1);
+ CountDownLatch release = new CountDownLatch(1);
+ Thread thread = new Thread(() -> {
+ started.countDown();
+ try {
+ release.await(30, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }, "paimon-reader-async-thread-test");
+
+ thread.start();
+ try {
+ Assert.assertTrue(started.await(5, TimeUnit.SECONDS));
+
Assert.assertTrue(PaimonJniScanner.countThreadsByNamePrefix("paimon-reader-async-thread")
>= 1);
+ } finally {
+ release.countDown();
+ thread.join(5000);
+ }
+ }
+
+ @Test
+ public void testParseDataSizeBytes() {
+ Assert.assertEquals(Long.valueOf(1024L),
PaimonJniScanner.parseDataSizeBytes("1 KiB").get());
+ Assert.assertEquals(Long.valueOf(10L * 1024L * 1024L),
+ PaimonJniScanner.parseDataSizeBytes("10 MiB").get());
+ Assert.assertEquals(Long.valueOf(2L * 1024L * 1024L * 1024L),
+ PaimonJniScanner.parseDataSizeBytes("2GB").get());
+
Assert.assertFalse(PaimonJniScanner.parseDataSizeBytes("unknown").isPresent());
+ }
+
private Map<String, String> createBaseParams() {
Map<String, String> params = new HashMap<>();
params.put("required_fields", "");
@@ -102,6 +168,22 @@ public class PaimonJniScannerTest {
return params;
}
+ private void setTableOptions(PaimonJniScanner scanner, Map<String, String>
options) throws Exception {
+ Table table = (Table) Proxy.newProxyInstance(
+ Table.class.getClassLoader(), new Class[] {Table.class},
(proxy, method, args) -> {
+ if ("options".equals(method.getName())) {
+ return options;
+ }
+ if ("toString".equals(method.getName())) {
+ return "TestPaimonTable";
+ }
+ throw new UnsupportedOperationException(method.getName());
+ });
+ Field tableField = PaimonJniScanner.class.getDeclaredField("table");
+ tableField.setAccessible(true);
+ tableField.set(scanner, table);
+ }
+
public static class TestIOManager implements IOManager {
private final String[] tempDirs;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
