github-actions[bot] commented on code in PR #65354:
URL: https://github.com/apache/doris/pull/65354#discussion_r3541448517
##########
fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonJniScanner.java:
##########
@@ -363,10 +405,159 @@ protected TableSchema parseTableSchema() throws
UnsupportedOperationException {
@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("peak:PaimonJniActiveScannerPeakCount",
String.valueOf(PEAK_ACTIVE_SCANNERS.get()));
+ statistics.put("gauge:PaimonJniAsyncReaderThreadCount",
+ String.valueOf(currentAsyncReaderThreadCount()));
+ statistics.put("peak:PaimonJniAsyncReaderThreadPeakCount",
+ String.valueOf(PEAK_ASYNC_READER_THREADS.get()));
+ 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() {
+ int currentCount =
countThreadsByNamePrefix(ASYNC_READER_THREAD_NAME_PREFIX);
+ updatePeak(PEAK_ASYNC_READER_THREADS, currentCount);
+ return currentCount;
+ }
+
+ static int countThreadsByNamePrefix(String threadNamePrefix) {
+ int count = 0;
+ for (Thread thread : Thread.getAllStackTraces().keySet()) {
+ if (thread.getName().startsWith(threadNamePrefix)) {
Review Comment:
`getStatistics()` now calls `currentAsyncReaderThreadCount()` for every
Paimon JNI scanner close, and this helper uses
`Thread.getAllStackTraces().keySet()` even though only thread names are needed.
That API builds stack traces for every live JVM thread, so a query with many
Paimon JNI splits will repeat a JVM-wide stack walk and allocate stack trace
arrays once per split in both the legacy and v2 collectors. This makes the new
observability path much heavier than intended, especially on BEs with many JVM
threads. Please avoid collecting full stack traces here, for example by using
`ThreadMXBean.getAllThreadIds()` plus `getThreadInfo(ids, 0)`/thread names, or
another lightweight way to track the Paimon async-reader executor thread count.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
