This is an automated email from the ASF dual-hosted git repository.
jerry-024 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-vector-index.git
The following commit(s) were added to refs/heads/main by this push:
new ee17bfd feat(logging): bridge native IVF-PQ diagnostics to SLF4J (#75)
ee17bfd is described below
commit ee17bfde5028e5d46aacc72192d078daab7eceae
Author: jerry <[email protected]>
AuthorDate: Thu Aug 13 09:38:58 2026 +0800
feat(logging): bridge native IVF-PQ diagnostics to SLF4J (#75)
---
.github/workflows/ci.yml | 8 +-
core/src/io.rs | 68 ++++-
core/src/ivfpq.rs | 325 ++++++++++++++++++++-
core/src/lib.rs | 1 +
core/src/logging.rs | 93 ++++++
core/src/sparse_table.rs | 8 +
java/pom.xml | 42 +++
.../paimon/index/vector/NativeLogBridge.java | 62 ++++
.../index/vector/NativeLogBridgeSmokeTest.java | 202 +++++++++++++
jni/src/lib.rs | 1 +
jni/src/log_bridge.rs | 130 +++++++++
11 files changed, 925 insertions(+), 15 deletions(-)
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index e8b61d8..edeae96 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -191,12 +191,14 @@ jobs:
java-version: '8'
distribution: 'temurin'
- - name: Test Java API
- run: mvn -f java/pom.xml test
-
- name: Build JNI library
run: cargo build -p paimon-vindex-jni --release
+ - name: Test Java API
+ run: >
+ mvn -f java/pom.xml test
+ -Dpaimon.vindex.native.path="${{ github.workspace
}}/target/release/libpaimon_vindex_jni.so"
+
- name: Test JNI native behavior
run: |
java -cp java/target/test-classes:java/target/classes
org.apache.paimon.index.vector.VectorIndexNativeValidationTest \
diff --git a/core/src/io.rs b/core/src/io.rs
index 87b7dd3..29fa69a 100644
--- a/core/src/io.rs
+++ b/core/src/io.rs
@@ -27,6 +27,7 @@ use crate::pq::ProductQuantizer;
use rayon::prelude::*;
use std::io;
use std::mem::size_of;
+use std::time::{Duration, Instant};
pub const MAGIC: u32 = 0x49565051; // "IVPQ"
pub const VERSION: u32 = 1;
@@ -95,6 +96,56 @@ pub trait SeekRead: Send {
}
}
+#[derive(Clone, Copy, Debug, Default)]
+pub(crate) struct ReadMetrics {
+ pub elapsed: Duration,
+ pub calls: usize,
+ pub requested_bytes: usize,
+}
+
+struct MeasuredSeekRead<R> {
+ inner: R,
+ metrics: Option<ReadMetrics>,
+}
+
+impl<R> MeasuredSeekRead<R> {
+ fn new(inner: R) -> Self {
+ Self {
+ inner,
+ metrics: None,
+ }
+ }
+}
+
+impl<R: SeekRead> SeekRead for MeasuredSeekRead<R> {
+ fn pread(&mut self, ranges: &mut [ReadRequest<'_>]) -> io::Result<()> {
+ let measurement = self.metrics.as_ref().map(|_| {
+ let requested_bytes = ranges.iter().fold(0usize, |total, request| {
+ total.saturating_add(request.buf.len())
+ });
+ (Instant::now(), requested_bytes)
+ });
+ let result = self.inner.pread(ranges);
+ if let (Some(metrics), Some((started, requested_bytes))) =
+ (self.metrics.as_mut(), measurement)
+ {
+ metrics.elapsed += started.elapsed();
+ metrics.calls = metrics.calls.saturating_add(1);
+ metrics.requested_bytes =
metrics.requested_bytes.saturating_add(requested_bytes);
+ }
+ result
+ }
+
+ fn try_clone_reader(&self) -> io::Result<Option<Self>> {
+ // Clones start with metrics disabled, so their I/O is not included
here.
+ Ok(self.inner.try_clone_reader()?.map(Self::new))
+ }
+
+ fn read_capabilities(&self) -> SeekReadCapabilities {
+ self.inner.read_capabilities()
+ }
+}
+
pub(crate) struct PreadCursor<'a, R: SeekRead + ?Sized> {
reader: &'a mut R,
pos: u64,
@@ -456,7 +507,7 @@ fn u64_to_i64(value: u64, field: &str) -> io::Result<i64> {
// --- Reader ---
pub struct IVFPQIndexReader<R: SeekRead> {
- reader: R,
+ reader: MeasuredSeekRead<R>,
pub d: usize,
pub nlist: usize,
pub m: usize,
@@ -578,7 +629,7 @@ impl<R: SeekRead> IVFPQIndexReader<R> {
let has_opq = flags & FLAG_HAS_OPQ != 0;
Ok(IVFPQIndexReader {
- reader,
+ reader: MeasuredSeekRead::new(reader),
d,
nlist,
m,
@@ -609,6 +660,14 @@ impl<R: SeekRead> IVFPQIndexReader<R> {
})
}
+ pub(crate) fn begin_read_metrics(&mut self) {
+ self.reader.metrics = Some(ReadMetrics::default());
+ }
+
+ pub(crate) fn end_read_metrics(&mut self) -> ReadMetrics {
+ self.reader.metrics.take().unwrap_or_default()
+ }
+
/// Load centroids, codebooks, and offset table. Called automatically on
first search.
pub fn ensure_loaded(&mut self) -> io::Result<()> {
if self.loaded {
@@ -1396,9 +1455,12 @@ mod tests {
reader.list_id_bytes_lens[non_empty_list] > 0,
"v1 files must store id_bytes_len in the offset table"
);
+ let expected_requested_bytes =
reader.list_payload_len(non_empty_list).unwrap();
+ reader.begin_read_metrics();
let mut lists = reader
.read_inverted_list_payloads(&[non_empty_list])
.unwrap();
+ let read_metrics = reader.end_read_metrics();
let list = lists.pop().unwrap();
let read_ids = &list.ids;
let codes = list.codes();
@@ -1416,6 +1478,8 @@ mod tests {
stats.pread_calls, 1,
"delta-varint lists with offset-table id length should use one
pread"
);
+ assert_eq!(read_metrics.calls, 1);
+ assert_eq!(read_metrics.requested_bytes, expected_requested_bytes);
}
#[test]
diff --git a/core/src/ivfpq.rs b/core/src/ivfpq.rs
index 7cfd70f..e45647f 100644
--- a/core/src/ivfpq.rs
+++ b/core/src/ivfpq.rs
@@ -22,14 +22,17 @@ use crate::distance::{
use crate::index_io_util::ivf_payload_is_oversized;
use crate::io::{IVFPQIndexReader, InvertedListPayload, SeekRead};
use crate::kmeans::{self, KMeansConfig};
+use crate::logging::{emit_log, LogLevel};
use crate::opq::OPQMatrix;
use crate::pq::ProductQuantizer;
+use crate::sparse_table::SparseTable;
use rayon::prelude::*;
use roaring::RoaringTreemap;
use std::borrow::Cow;
use std::collections::HashSet;
use std::io;
use std::sync::OnceLock;
+use std::time::{Duration, Instant};
pub trait RowIdFilter: Sync {
fn contains(&self, id: i64) -> bool;
@@ -1722,6 +1725,131 @@ fn scan_reader_codes(
}
}
+#[derive(Default)]
+struct IvfpqBatchTiming {
+ load: Duration,
+ preprocess: Duration,
+ coarse: Duration,
+ prepare: Duration,
+ io_read: Duration,
+ decode: Duration,
+ filter: Duration,
+ scan: Duration,
+ finalize: Duration,
+ read_calls: usize,
+ requested_bytes: usize,
+ unique_list_rows: usize,
+ query_list_pairs: usize,
+ pq_codes_evaluated: usize,
+ sparse_query_list_pairs: usize,
+ dense_query_list_pairs: usize,
+ actual_pq_codes_evaluated: usize,
+ matched_rows: usize,
+ queries_below_k: usize,
+ min_hits_per_query: usize,
+}
+
+impl IvfpqBatchTiming {
+ fn record_scan_work(
+ &mut self,
+ rows: usize,
+ matching_rows: Option<&MatchingRows>,
+ query_uses: usize,
+ pq_bits: usize,
+ transposed_codes: bool,
+ ) {
+ let matched_rows = matching_rows.map_or(rows, MatchingRows::len);
+ self.unique_list_rows = self.unique_list_rows.saturating_add(rows);
+ self.matched_rows = self.matched_rows.saturating_add(matched_rows);
+ self.pq_codes_evaluated = self
+ .pq_codes_evaluated
+ .saturating_add(matched_rows.saturating_mul(query_uses));
+
+ if matched_rows == 0 || query_uses == 0 {
+ return;
+ }
+ let sparse = match (pq_bits, transposed_codes, matching_rows) {
+ (4, _, _) | (_, _, None) => false,
+ (_, true, Some(matching_rows)) => {
+ should_scan_sparse(rows, matching_rows,
TRANSPOSED_SPARSE_SCAN_DIVISOR)
+ }
+ (_, false, Some(matching_rows)) => {
+ should_scan_sparse(rows, matching_rows,
ROW_MAJOR_SPARSE_SCAN_DIVISOR)
+ }
+ };
+ let scan_rows = if sparse { matched_rows } else { rows };
+ self.actual_pq_codes_evaluated = self
+ .actual_pq_codes_evaluated
+ .saturating_add(scan_rows.saturating_mul(query_uses));
+ if sparse {
+ self.sparse_query_list_pairs =
self.sparse_query_list_pairs.saturating_add(query_uses);
+ } else {
+ self.dense_query_list_pairs =
self.dense_query_list_pairs.saturating_add(query_uses);
+ }
+ }
+
+ fn write_to<W: io::Write>(
+ &self,
+ mut output: W,
+ total: Duration,
+ nq: usize,
+ nprobe: usize,
+ pq_bits: usize,
+ topk: usize,
+ unique_lists: usize,
+ filtered: bool,
+ ) -> io::Result<()> {
+ let millis = |duration: Duration| duration.as_secs_f64() * 1_000.0;
+ writeln!(
+ output,
+ "[paimon-vindex] ivfpq_batch_timing nq={nq} nprobe={nprobe}
pq_bits={pq_bits} \
+ topk={topk} unique_lists={unique_lists} unique_list_rows={}
query_list_pairs={} \
+ pq_codes_evaluated={} sparse_query_list_pairs={}
dense_query_list_pairs={} \
+ actual_pq_codes_evaluated={} matched_rows={} read_calls={}
requested_bytes={} \
+ queries_below_k={} min_hits_per_query={} filtered={filtered}
total_ms={:.3} \
+ load_ms={:.3} preprocess_ms={:.3} coarse_ms={:.3}
prepare_ms={:.3} \
+ io_read_ms={:.3} decode_ms={:.3} filter_ms={:.3} scan_ms={:.3}
finalize_ms={:.3}",
+ self.unique_list_rows,
+ self.query_list_pairs,
+ self.pq_codes_evaluated,
+ self.sparse_query_list_pairs,
+ self.dense_query_list_pairs,
+ self.actual_pq_codes_evaluated,
+ self.matched_rows,
+ self.read_calls,
+ self.requested_bytes,
+ self.queries_below_k,
+ self.min_hits_per_query,
+ millis(total),
+ millis(self.load),
+ millis(self.preprocess),
+ millis(self.coarse),
+ millis(self.prepare),
+ millis(self.io_read),
+ millis(self.decode),
+ millis(self.filter),
+ millis(self.scan),
+ millis(self.finalize),
+ )
+ }
+}
+
+#[inline]
+fn elapsed_since(started: Option<Instant>) -> Duration {
+ started.map_or(Duration::ZERO, |started| started.elapsed())
+}
+
+fn end_read_metrics_on_error<R: SeekRead, T>(
+ reader: &mut IVFPQIndexReader<R>,
+ result: io::Result<T>,
+ timing_enabled: bool,
+) -> io::Result<T> {
+ if timing_enabled && result.is_err() {
+ let _ = reader.end_read_metrics();
+ }
+ result
+}
+
/// Big batch search: batch queries share list reads.
/// Instead of nq*nprobe I/O ops, reads each unique list once and scans for
all queries.
pub fn search_batch_reader<R: SeekRead>(
@@ -1949,7 +2077,12 @@ fn
search_batch_reader_filter_with_reuse_mode_and_observer<R: SeekRead>(
mut observe_ephemeral_precomputed_lists: impl FnMut(usize),
#[cfg(test)] distance_table_builds:
Option<&std::sync::atomic::AtomicUsize>,
) -> io::Result<(Vec<i64>, Vec<f32>)> {
+ let timing_enabled =
std::env::var_os("PAIMON_VINDEX_LOG_IVFPQ_BATCH_TIMING").is_some();
+ let total_started = timing_enabled.then(Instant::now);
+ let mut timing = IvfpqBatchTiming::default();
+ let load_started = timing_enabled.then(Instant::now);
reader.ensure_loaded()?;
+ timing.load = elapsed_since(load_started);
let d = reader.d;
if nq == 0 {
return Err(io::Error::new(
@@ -1985,6 +2118,7 @@ fn
search_batch_reader_filter_with_reuse_mode_and_observer<R: SeekRead>(
"probe range must be non-empty",
));
}
+ let scanned_nprobe = probe_end - probe_start;
validate_batch_seed(seed_ids, seed_distances, nq, k)?;
let m = reader.m;
@@ -1993,6 +2127,7 @@ fn
search_batch_reader_filter_with_reuse_mode_and_observer<R: SeekRead>(
let by_residual = reader.by_residual;
// Step 1: Preprocess all queries
+ let preprocess_started = timing_enabled.then(Instant::now);
let mut processed = queries[..nq * d].to_vec();
if metric == MetricType::Cosine {
for i in 0..nq {
@@ -2004,8 +2139,10 @@ fn
search_batch_reader_filter_with_reuse_mode_and_observer<R: SeekRead>(
opq.apply_batch(&processed, &mut rotated, nq);
processed = rotated;
}
+ timing.preprocess = elapsed_since(preprocess_started);
// Step 2: Batch coarse search (one sgemm)
+ let coarse_started = timing_enabled.then(Instant::now);
let (all_probe_indices, all_coarse_dists) = kmeans::find_topk_batch(
&processed,
nq,
@@ -2014,13 +2151,28 @@ fn
search_batch_reader_filter_with_reuse_mode_and_observer<R: SeekRead>(
d,
probe_end,
);
+ timing.coarse = elapsed_since(coarse_started);
// Step 3: Read every probed list once. Queries share the decoded list
// payloads, then scan independently in parallel.
+ let prepare_started = timing_enabled.then(Instant::now);
let mut seen = vec![false; reader.nlist];
let mut unique_lists = Vec::new();
+ let mut query_uses_by_list = timing_enabled
+ .then(||
SparseTable::<usize>::with_capacity(scanned_nprobe.min(reader.nlist)));
for probe_indices in &all_probe_indices {
for &list_id in probe_indices.iter().skip(probe_start) {
+ if let Some(query_uses) = query_uses_by_list.as_mut() {
+ timing.query_list_pairs =
timing.query_list_pairs.saturating_add(1);
+ if reader.list_counts[list_id] > 0 {
+ let key = list_id as u32;
+ if let Some(uses) = query_uses.get_mut(key) {
+ *uses = uses.saturating_add(1);
+ } else {
+ let _ = query_uses.insert(key, 1);
+ }
+ }
+ }
if !seen[list_id] && reader.list_counts[list_id] > 0 {
seen[list_id] = true;
unique_lists.push(list_id);
@@ -2094,13 +2246,20 @@ fn
search_batch_reader_filter_with_reuse_mode_and_observer<R: SeekRead>(
Vec::new()
};
let mut stable_pq_norms = None;
+ timing.prepare = elapsed_since(prepare_started);
let mut heaps = (0..nq).map(|_| TopKHeap::new(k)).collect::<Vec<_>>();
seed_heaps(&mut heaps, seed_ids, seed_distances, k);
+ if timing_enabled {
+ reader.begin_read_metrics();
+ }
let mut batch_start = 0usize;
while batch_start < unique_lists.len() {
let first_list = unique_lists[batch_start];
- if ivf_payload_is_oversized(reader.list_payload_len(first_list)?) {
+ let payload_len = reader.list_payload_len(first_list);
+ let payload_len = end_read_metrics_on_error(reader, payload_len,
timing_enabled)?;
+ if ivf_payload_is_oversized(payload_len) {
+ let prepare_started = timing_enabled.then(Instant::now);
let query_tables = (0..nq)
.filter_map(|query_index| {
all_probe_indices[query_index]
@@ -2139,11 +2298,27 @@ fn
search_batch_reader_filter_with_reuse_mode_and_observer<R: SeekRead>(
// The loop is sequential across queries. Reuse one chunk-sized
// distance buffer instead of retaining one per query.
let mut distances = Vec::new();
- reader.for_each_streamed_list_chunk(first_list, |pq, ids, codes| {
+ timing.prepare += elapsed_since(prepare_started);
+ let streamed_started = timing_enabled.then(Instant::now);
+ let mut streamed_filter = Duration::ZERO;
+ let mut streamed_scan = Duration::ZERO;
+ let result = reader.for_each_streamed_list_chunk(first_list, |pq,
ids, codes| {
+ let filter_started = timing_enabled.then(Instant::now);
let positions = matching_rows(ids, filter);
+ streamed_filter += elapsed_since(filter_started);
+ if timing_enabled {
+ timing.record_scan_work(
+ ids.len(),
+ positions.as_ref(),
+ query_tables.len(),
+ pq_nbits,
+ transposed_codes,
+ );
+ }
if positions.as_ref().is_some_and(MatchingRows::is_empty) {
return;
}
+ let scan_started = timing_enabled.then(Instant::now);
for (query_index, dis0, sim_table) in &query_tables {
let sim_table = sim_table.as_deref().unwrap_or_else(|| {
shared_sim_tables[*query_index].get_or_init(|| {
@@ -2174,22 +2349,52 @@ fn
search_batch_reader_filter_with_reuse_mode_and_observer<R: SeekRead>(
&mut heaps[*query_index],
);
}
- })?;
+ streamed_scan += elapsed_since(scan_started);
+ });
+ end_read_metrics_on_error(reader, result, timing_enabled)?;
+ let streamed_total = elapsed_since(streamed_started);
+ timing.filter += streamed_filter;
+ timing.scan += streamed_scan;
+ // Keep I/O plus decode here; measured I/O is removed below for
both read paths.
+ timing.decode += streamed_total.saturating_sub(streamed_filter +
streamed_scan);
batch_start += 1;
continue;
}
- let count =
reader.batch_read_end(&unique_lists[batch_start..])?.max(1);
+ let batch_end_result =
reader.batch_read_end(&unique_lists[batch_start..]);
+ let count = end_read_metrics_on_error(reader, batch_end_result,
timing_enabled)?.max(1);
let batch_end = (batch_start + count).min(unique_lists.len());
- let loaded_lists =
-
reader.read_inverted_list_payloads(&unique_lists[batch_start..batch_end])?;
+ let read_decode_started = timing_enabled.then(Instant::now);
+ let loaded_lists_result =
+
reader.read_inverted_list_payloads(&unique_lists[batch_start..batch_end]);
+ let loaded_lists = end_read_metrics_on_error(reader,
loaded_lists_result, timing_enabled)?;
+ timing.decode += elapsed_since(read_decode_started);
+ let prepare_started = timing_enabled.then(Instant::now);
let mut list_positions = vec![usize::MAX; reader.nlist];
for (position, list) in loaded_lists.iter().enumerate() {
list_positions[list.list_id] = position;
}
+ timing.prepare += elapsed_since(prepare_started);
+ let filter_started = timing_enabled.then(Instant::now);
let matching_rows_by_list = loaded_lists
.iter()
.map(|list| matching_rows(&list.ids, filter))
.collect::<Vec<_>>();
+ timing.filter += elapsed_since(filter_started);
+ if let Some(query_uses) = query_uses_by_list.as_ref() {
+ for (list, matching_rows) in
loaded_lists.iter().zip(&matching_rows_by_list) {
+ timing.record_scan_work(
+ list.ids.len(),
+ matching_rows.as_ref(),
+ query_uses
+ .get(list.list_id as u32)
+ .copied()
+ .unwrap_or_default(),
+ reader.pq.nbits,
+ reader.transposed_codes,
+ );
+ }
+ }
+ let scan_started = timing_enabled.then(Instant::now);
let matching_list_count = matching_rows_by_list
.iter()
.filter(|rows| has_matching_rows(rows.as_ref()))
@@ -2354,29 +2559,59 @@ fn
search_batch_reader_filter_with_reuse_mode_and_observer<R: SeekRead>(
heaps[qi].push(distance, row_id);
}
}
+ timing.scan += elapsed_since(scan_started);
batch_start = batch_end;
}
+ if timing_enabled {
+ let read_metrics = reader.end_read_metrics();
+ timing.io_read = read_metrics.elapsed;
+ // Both batch and streamed reads accumulated I/O plus decode above.
+ timing.decode = timing.decode.saturating_sub(timing.io_read);
+ timing.read_calls = read_metrics.calls;
+ timing.requested_bytes = read_metrics.requested_bytes;
+ }
+ let finalize_started = timing_enabled.then(Instant::now);
let mut result_ids = vec![-1i64; nq * k];
let mut result_dists = vec![f32::MAX; nq * k];
+ timing.min_hits_per_query = k;
for (qi, heap) in heaps.into_iter().enumerate() {
let sorted = heap.into_sorted();
+ if timing_enabled {
+ timing.queries_below_k = timing
+ .queries_below_k
+ .saturating_add(usize::from(sorted.len() < k));
+ timing.min_hits_per_query =
timing.min_hits_per_query.min(sorted.len());
+ }
let base = qi * k;
for (i, &(dist, id)) in sorted.iter().enumerate() {
result_ids[base + i] = id;
result_dists[base + i] = dist;
}
}
+ timing.finalize = elapsed_since(finalize_started);
- if !by_residual &&
std::env::var_os("PAIMON_VINDEX_LOG_IVFPQ_BATCH_REUSE").is_some() {
- use std::io::Write;
+ if timing_enabled {
+ let mut buf = Vec::with_capacity(256);
+ let _ = timing.write_to(
+ &mut buf,
+ elapsed_since(total_started),
+ nq,
+ scanned_nprobe,
+ reader.pq.nbits,
+ k,
+ unique_lists.len(),
+ filter.is_some(),
+ );
+ emit_log(LogLevel::Info, String::from_utf8_lossy(&buf).trim_end());
+ }
+ if !by_residual &&
std::env::var_os("PAIMON_VINDEX_LOG_IVFPQ_BATCH_REUSE").is_some() {
let tables_built = shared_sim_tables
.iter()
.filter(|table| table.get().is_some())
.count();
- let _ = writeln!(
- std::io::stderr().lock(),
+ let message = format!(
"[paimon-vindex] ivfpq_batch_table_reuse
strategy=non_residual_query_table \
mode={reuse_mode:?} enabled={reuse_non_residual_tables} used={}
metric={} \
pq_bits={} nq={nq} nprobe={probe_end} unique_lists={} filtered={}
required_bytes={:?} \
@@ -2388,6 +2623,7 @@ fn
search_batch_reader_filter_with_reuse_mode_and_observer<R: SeekRead>(
filter.is_some(),
reuse_required_bytes,
);
+ emit_log(LogLevel::Info, &message);
}
Ok((result_ids, result_dists))
@@ -3338,6 +3574,75 @@ mod tests {
assert_eq!(actual, expected);
}
+ #[test]
+ fn ivfpq_batch_timing_output_names_search_phases() {
+ let timing = IvfpqBatchTiming {
+ load: std::time::Duration::from_millis(1),
+ preprocess: std::time::Duration::from_millis(2),
+ coarse: std::time::Duration::from_millis(3),
+ prepare: std::time::Duration::from_millis(4),
+ io_read: std::time::Duration::from_millis(5),
+ decode: std::time::Duration::from_millis(6),
+ filter: std::time::Duration::from_millis(7),
+ scan: std::time::Duration::from_millis(8),
+ finalize: std::time::Duration::from_millis(9),
+ read_calls: 10,
+ requested_bytes: 4096,
+ unique_list_rows: 120,
+ query_list_pairs: 512,
+ pq_codes_evaluated: 240,
+ sparse_query_list_pairs: 128,
+ dense_query_list_pairs: 384,
+ actual_pq_codes_evaluated: 960,
+ matched_rows: 30,
+ queries_below_k: 2,
+ min_hits_per_query: 1,
+ };
+ let mut output = Vec::new();
+
+ timing
+ .write_to(
+ &mut output,
+ std::time::Duration::from_millis(45),
+ 64,
+ 8,
+ 4,
+ 3,
+ 12,
+ true,
+ )
+ .unwrap();
+
+ assert_eq!(
+ String::from_utf8(output).unwrap(),
+ "[paimon-vindex] ivfpq_batch_timing nq=64 nprobe=8 pq_bits=4 \
+ topk=3 unique_lists=12 unique_list_rows=120 query_list_pairs=512 \
+ pq_codes_evaluated=240 sparse_query_list_pairs=128
dense_query_list_pairs=384 \
+ actual_pq_codes_evaluated=960 matched_rows=30 read_calls=10
requested_bytes=4096 \
+ queries_below_k=2 min_hits_per_query=1 filtered=true
total_ms=45.000 load_ms=1.000 \
+ preprocess_ms=2.000 coarse_ms=3.000 prepare_ms=4.000 \
+ io_read_ms=5.000 decode_ms=6.000 filter_ms=7.000 scan_ms=8.000 \
+ finalize_ms=9.000\n"
+ );
+ }
+
+ #[test]
+ fn ivfpq_batch_timing_distinguishes_sparse_and_dense_scan_work() {
+ let mut timing = IvfpqBatchTiming::default();
+ let sparse_rows = MatchingRows::Sparse((0..10).collect());
+ let dense_rows = MatchingRows::Sparse((0..20).collect());
+
+ timing.record_scan_work(100, Some(&sparse_rows), 4, 8, true);
+ timing.record_scan_work(100, Some(&dense_rows), 3, 8, true);
+
+ assert_eq!(timing.unique_list_rows, 200);
+ assert_eq!(timing.matched_rows, 30);
+ assert_eq!(timing.pq_codes_evaluated, 100);
+ assert_eq!(timing.sparse_query_list_pairs, 4);
+ assert_eq!(timing.dense_query_list_pairs, 3);
+ assert_eq!(timing.actual_pq_codes_evaluated, 340);
+ }
+
#[test]
fn matching_rows_adapts_sparse_positions_to_bounded_bitmap() {
let ids = (0..1024i64).collect::<Vec<_>>();
diff --git a/core/src/lib.rs b/core/src/lib.rs
index da0c188..6d9c64c 100644
--- a/core/src/lib.rs
+++ b/core/src/lib.rs
@@ -36,6 +36,7 @@ pub mod ivfrq_io;
pub mod ivfsq;
pub mod ivfsq_io;
pub mod kmeans;
+pub mod logging;
pub mod opq;
pub mod pq;
pub mod read_options;
diff --git a/core/src/logging.rs b/core/src/logging.rs
new file mode 100644
index 0000000..87f2934
--- /dev/null
+++ b/core/src/logging.rs
@@ -0,0 +1,93 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Pluggable diagnostic log sink.
+//!
+//! By default runtime diagnostics are written to stderr, which keeps the
+//! historical behavior for the C FFI and Python consumers. Embedders (such as
+//! the JNI layer) may install a process-wide sink once to redirect records to
+//! their own logging system (e.g. SLF4J/log4j in Spark executors).
+
+use std::io::Write;
+use std::sync::OnceLock;
+
+/// Severity of a diagnostic record emitted by the core library.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum LogLevel {
+ Error = 1,
+ Warn = 2,
+ Info = 3,
+ Debug = 4,
+}
+
+/// Process-wide log sink. Receives the level and a message without a trailing
+/// newline. Implementations must never panic.
+pub type LogSink = Box<dyn Fn(LogLevel, &str) + Send + Sync>;
+
+static LOG_SINK: OnceLock<LogSink> = OnceLock::new();
+
+/// Installs the process-wide sink. The first caller wins; returns
+/// `Err(sink)` if a sink is already installed.
+pub fn set_log_sink(sink: LogSink) -> Result<(), LogSink> {
+ LOG_SINK.set(sink)
+}
+
+/// Emits one record through the installed sink, or falls back to stderr.
+pub(crate) fn emit_log(level: LogLevel, message: &str) {
+ match LOG_SINK.get() {
+ Some(sink) => sink(level, message),
+ None => {
+ let _ = writeln!(std::io::stderr().lock(), "{message}");
+ }
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use std::sync::{Arc, Mutex};
+
+ // The sink is a process-global OnceLock shared by every test in this
+ // binary, so install/deliver/reject must run inside one test.
+ #[test]
+ fn sink_install_deliver_and_reject_second_install() {
+ let captured: Arc<Mutex<Vec<(LogLevel, String)>>> =
Arc::new(Mutex::new(Vec::new()));
+ let sink_capture = Arc::clone(&captured);
+ set_log_sink(Box::new(move |level, message| {
+ sink_capture
+ .lock()
+ .unwrap()
+ .push((level, message.to_string()));
+ }))
+ .unwrap_or_else(|_| panic!("first install must succeed"));
+
+ emit_log(LogLevel::Info, "hello");
+ emit_log(LogLevel::Warn, "watch out");
+
+ let records = captured.lock().unwrap();
+ assert_eq!(
+ *records,
+ vec![
+ (LogLevel::Info, "hello".to_string()),
+ (LogLevel::Warn, "watch out".to_string()),
+ ]
+ );
+ drop(records);
+
+ assert!(set_log_sink(Box::new(|_, _| {})).is_err());
+ }
+}
diff --git a/core/src/sparse_table.rs b/core/src/sparse_table.rs
index 11072d8..6929d53 100644
--- a/core/src/sparse_table.rs
+++ b/core/src/sparse_table.rs
@@ -69,6 +69,12 @@ impl<V: Copy + Default> SparseTable<V> {
(self.keys[slot] == key).then(|| &self.values[slot])
}
+ pub(crate) fn get_mut(&mut self, key: u32) -> Option<&mut V> {
+ assert_ne!(key, EMPTY_KEY, "u32::MAX is reserved by SparseTable");
+ let slot = self.find_slot(key)?;
+ (self.keys[slot] == key).then(|| &mut self.values[slot])
+ }
+
pub(crate) fn insert(&mut self, key: u32, value: V) -> Option<V> {
assert_ne!(key, EMPTY_KEY, "u32::MAX is reserved by SparseTable");
let slot = self.find_slot(key).expect("SparseTable must have a slot");
@@ -171,6 +177,8 @@ mod tests {
assert_eq!(table.get(7), Some(&1));
assert_eq!(table.insert(7, 3), Some(1));
assert_eq!(table.get(7), Some(&3));
+ *table.get_mut(7).unwrap() = 5;
+ assert_eq!(table.get(7), Some(&5));
assert_eq!(table.len(), 1);
}
diff --git a/java/pom.xml b/java/pom.xml
index 6d0e1a4..0f48abf 100644
--- a/java/pom.xml
+++ b/java/pom.xml
@@ -67,8 +67,26 @@
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<!-- Used by maven-remote-resources-plugin when generating
META-INF/NOTICE. -->
<project.build.outputTimestamp>2026-01-01T00:00:00Z</project.build.outputTimestamp>
+ <!-- Optional external JNI library path forwarded to forked test JVMs;
+ empty means load the platform library bundled in resources. -->
+ <paimon.vindex.native.path></paimon.vindex.native.path>
</properties>
+ <dependencies>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-api</artifactId>
+ <version>1.7.36</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-simple</artifactId>
+ <version>1.7.36</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+
<build>
<resources>
<resource>
@@ -109,6 +127,30 @@
<classpathScope>test</classpathScope>
</configuration>
</execution>
+ <!-- Forked JVM (exec goal): the Rust timing gate reads
the process
+ environment, which the in-process java goal cannot
set. -->
+ <execution>
+ <id>native-log-bridge-smoke-test</id>
+ <phase>test</phase>
+ <goals>
+ <goal>exec</goal>
+ </goals>
+ <configuration>
+ <skip>${skipTests}</skip>
+ <executable>java</executable>
+ <classpathScope>test</classpathScope>
+ <environmentVariables>
+
<PAIMON_VINDEX_LOG_IVFPQ_BATCH_TIMING>1</PAIMON_VINDEX_LOG_IVFPQ_BATCH_TIMING>
+ </environmentVariables>
+ <arguments>
+
<argument>-Dvindex.smoke.require-timing=true</argument>
+
<argument>-Dpaimon.vindex.native.path=${paimon.vindex.native.path}</argument>
+ <argument>-classpath</argument>
+ <classpath/>
+
<argument>org.apache.paimon.index.vector.NativeLogBridgeSmokeTest</argument>
+ </arguments>
+ </configuration>
+ </execution>
</executions>
</plugin>
</plugins>
diff --git
a/java/src/main/java/org/apache/paimon/index/vector/NativeLogBridge.java
b/java/src/main/java/org/apache/paimon/index/vector/NativeLogBridge.java
new file mode 100644
index 0000000..fdf0fbe
--- /dev/null
+++ b/java/src/main/java/org/apache/paimon/index/vector/NativeLogBridge.java
@@ -0,0 +1,62 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.paimon.index.vector;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Receives diagnostic records from the Rust core (installed by JNI_OnLoad in
+ * jni/src/log_bridge.rs) and forwards them to SLF4J. Package-private is
+ * sufficient: JNI method lookup does not enforce Java access control.
+ */
+final class NativeLogBridge {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(NativeLogBridge.class);
+
+ private NativeLogBridge() {}
+
+ // Called from native code; the signature must remain
(ILjava/lang/String;)V.
+ // Must never throw: a pending exception here would surface on arbitrary
+ // native threads.
+ static void log(int level, String message) {
+ try {
+ switch (level) {
+ case 1:
+ LOG.error(message);
+ break;
+ case 2:
+ LOG.warn(message);
+ break;
+ case 4:
+ LOG.debug(message);
+ break;
+ case 3:
+ default:
+ LOG.info(message);
+ break;
+ }
+ } catch (Throwable ignored) {
+ try {
+ System.err.println(message);
+ } catch (Throwable ignoredFallback) {
+ // Keep the JNI callback non-throwing even if stderr itself
fails.
+ }
+ }
+ }
+}
diff --git
a/java/src/test/java/org/apache/paimon/index/vector/NativeLogBridgeSmokeTest.java
b/java/src/test/java/org/apache/paimon/index/vector/NativeLogBridgeSmokeTest.java
new file mode 100644
index 0000000..c774eae
--- /dev/null
+++
b/java/src/test/java/org/apache/paimon/index/vector/NativeLogBridgeSmokeTest.java
@@ -0,0 +1,202 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.paimon.index.vector;
+
+import java.io.ByteArrayOutputStream;
+import java.io.PrintStream;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Random;
+
+/**
+ * Standalone check that native IVF-PQ timing diagnostics reach SLF4J (via
+ * NativeLogBridge) instead of the raw process stderr.
+ *
+ * <p>Requires the environment variable PAIMON_VINDEX_LOG_IVFPQ_BATCH_TIMING
to be set
+ * before the JVM starts (the Rust gate reads the process environment); prints
a
+ * skip notice and exits 0 otherwise. Run with slf4j-simple on the classpath:
+ *
+ * <pre>
+ * PAIMON_VINDEX_LOG_IVFPQ_BATCH_TIMING=1 java -cp ... \
+ * org.apache.paimon.index.vector.NativeLogBridgeSmokeTest
[/path/to/libpaimon_vindex_jni.so]
+ * </pre>
+ */
+public class NativeLogBridgeSmokeTest {
+
+ private static final String TIMING_MARKER = "ivfpq_batch_timing";
+ private static final String SLF4J_TIMING_PREFIX =
+ "INFO "
+ + NativeLogBridge.class.getName()
+ + " - [paimon-vindex] "
+ + TIMING_MARKER;
+ private static final String[] REQUIRED_TIMING_FIELDS = {
+ "topk",
+ "unique_list_rows",
+ "query_list_pairs",
+ "pq_codes_evaluated",
+ "matched_rows",
+ "read_calls",
+ "requested_bytes",
+ "queries_below_k",
+ "min_hits_per_query",
+ "io_read_ms",
+ "decode_ms"
+ };
+
+ public static void main(String[] args) {
+ if (System.getenv("PAIMON_VINDEX_LOG_IVFPQ_BATCH_TIMING") == null) {
+ if (Boolean.getBoolean("vindex.smoke.require-timing")) {
+ throw new AssertionError(
+ "PAIMON_VINDEX_LOG_IVFPQ_BATCH_TIMING must be set in
the process environment "
+ + "when vindex.smoke.require-timing=true");
+ }
+ System.out.println(
+ "SKIP: PAIMON_VINDEX_LOG_IVFPQ_BATCH_TIMING is not set in
the process environment");
+ return;
+ }
+ VectorIndexNativeLoaderSmokeTest.configureExternalLibrary(args);
+ if (!hasNativeLibrary()) {
+ System.out.println("SKIP: no explicit or bundled JNI library is
available");
+ return;
+ }
+
+ PrintStream originalOut = System.out;
+ PrintStream originalErr = System.err;
+ ByteArrayOutputStream capturedOut = new ByteArrayOutputStream();
+ ByteArrayOutputStream capturedErr = new ByteArrayOutputStream();
+ String out;
+ String err;
+ try {
+ // Capture before the first native/SLF4J use: slf4j-simple logs to
+ // System.err by default and does not cache the stream.
+ System.setOut(new PrintStream(capturedOut, true));
+ System.setErr(new PrintStream(capturedErr, true));
+ runIvfPqBatchSearch();
+ } finally {
+ System.setOut(originalOut);
+ System.setErr(originalErr);
+ out = capturedOut.toString();
+ err = capturedErr.toString();
+ }
+
+ if (out.contains(TIMING_MARKER)) {
+ throw new AssertionError(
+ "timing record leaked to stdout instead of the log
bridge:\n" + out);
+ }
+ if (!err.contains(SLF4J_TIMING_PREFIX)) {
+ throw new AssertionError(
+ "timing record missing from SLF4J output.\nstderr:\n"
+ + err
+ + "\nstdout:\n"
+ + out);
+ }
+ for (String field : REQUIRED_TIMING_FIELDS) {
+ if (!err.contains(field + "=")) {
+ throw new AssertionError("timing field " + field + "
missing:\n" + err);
+ }
+ }
+ if (err.contains("read_decode_ms=")) {
+ throw new AssertionError("combined read/decode timing was not
split:\n" + err);
+ }
+ if (err.contains("native log bridge disabled")) {
+ throw new AssertionError("bridge unexpectedly degraded to
stderr:\n" + err);
+ }
+ System.out.println("OK: " + TIMING_MARKER + " delivered via
NativeLogBridge/SLF4J only");
+ }
+
+ private static boolean hasNativeLibrary() {
+ String explicitPath =
System.getProperty(NativeLibraryLoader.NATIVE_PATH_PROPERTY);
+ if (explicitPath != null && !explicitPath.trim().isEmpty()) {
+ return true;
+ }
+ try {
+ String resourcePath =
+ NativeLibraryLoader.resourcePath(
+ System.getProperty("os.name"),
System.getProperty("os.arch"));
+ return NativeLogBridgeSmokeTest.class.getResource(resourcePath) !=
null;
+ } catch (UnsatisfiedLinkError ignored) {
+ return false;
+ }
+ }
+
+ private static void runIvfPqBatchSearch() {
+ int dimension = 8;
+ int vectorCount = 512;
+ float[] data = new float[vectorCount * dimension];
+ long[] ids = new long[vectorCount];
+ Random random = new Random(13L);
+ for (int row = 0; row < vectorCount; row++) {
+ ids[row] = row;
+ for (int column = 0; column < dimension; column++) {
+ data[row * dimension + column] = (float) random.nextGaussian();
+ }
+ }
+
+ Map<String, String> options = new HashMap<String, String>();
+ options.put("index.type", "ivf_pq");
+ options.put("dimension", Integer.toString(dimension));
+ options.put("nlist", "2");
+ options.put("metric", "l2");
+ options.put("use-opq", "false");
+
+ VectorIndexWriter writer =
+ new VectorIndexWriter(VectorIndexTrainer.train(options, data,
vectorCount));
+ byte[] indexBytes;
+ try {
+ writer.addVectors(ids, data, vectorCount);
+ VectorIndexNativeHandleSafetyTest.ByteArrayPositionOutputStream
output =
+ new
VectorIndexNativeHandleSafetyTest.ByteArrayPositionOutputStream();
+ writer.writeIndex(output);
+ indexBytes = output.toByteArray();
+ } finally {
+ writer.close();
+ }
+
+ VectorIndexReader reader = new VectorIndexReader(new
ByteArrayInput(indexBytes));
+ try {
+ int queryCount = 4;
+ int topK = 3;
+ float[] queries = new float[queryCount * dimension];
+ for (int offset = 0; offset < queries.length; offset++) {
+ queries[offset] = (float) random.nextGaussian();
+ }
+ VectorSearchBatchResult result =
+ reader.searchBatch(queries, queryCount, new
VectorSearchParams(topK, 2));
+ if (result.ids().length != queryCount * topK) {
+ throw new AssertionError("unexpected batch result size: " +
result.ids().length);
+ }
+ } finally {
+ reader.close();
+ }
+ }
+
+ private static final class ByteArrayInput implements VectorIndexInput {
+ private final byte[] data;
+
+ ByteArrayInput(byte[] data) {
+ this.data = data;
+ }
+
+ @Override
+ public void pread(long[] positions, byte[][] buffers) {
+ for (int i = 0; i < positions.length; i++) {
+ System.arraycopy(data, (int) positions[i], buffers[i], 0,
buffers[i].length);
+ }
+ }
+ }
+}
diff --git a/jni/src/lib.rs b/jni/src/lib.rs
index db6eb38..e025cbd 100644
--- a/jni/src/lib.rs
+++ b/jni/src/lib.rs
@@ -15,6 +15,7 @@
// specific language governing permissions and limitations
// under the License.
+mod log_bridge;
mod stream;
use jni::objects::{JByteArray, JClass, JFloatArray, JLongArray, JObject,
JValue};
diff --git a/jni/src/log_bridge.rs b/jni/src/log_bridge.rs
new file mode 100644
index 0000000..79fd786
--- /dev/null
+++ b/jni/src/log_bridge.rs
@@ -0,0 +1,130 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Installs a core log sink that forwards diagnostic records to
+//! `org.apache.paimon.index.vector.NativeLogBridge`, so native output reaches
+//! SLF4J/log4j instead of the raw process stderr (which Spark's
+//! System.out-to-log4j redirect cannot capture).
+
+use jni::objects::{GlobalRef, JStaticMethodID, JValue};
+use jni::signature::{Primitive, ReturnType};
+use jni::sys::{jint, JNI_ERR, JNI_VERSION_1_8};
+use jni::{JNIEnv, JavaVM};
+use paimon_vindex_core::logging::{set_log_sink, LogLevel};
+use std::ffi::c_void;
+use std::panic::{catch_unwind, AssertUnwindSafe};
+
+const BRIDGE_CLASS: &str = "org/apache/paimon/index/vector/NativeLogBridge";
+
+struct LogBridge {
+ vm: JavaVM,
+ // Pins NativeLogBridge (and its classloader) so the method id stays valid.
+ class: GlobalRef,
+ log_method: JStaticMethodID,
+}
+
+/// Called synchronously by `System.load` (`NativeLibraryLoader.load`). Never
+/// leaves a pending exception and never fails the library load just because
+/// logging is unavailable (e.g. slf4j-api absent outside Spark).
+#[no_mangle]
+pub extern "system" fn JNI_OnLoad(vm: *mut jni::sys::JavaVM, _reserved: *mut
c_void) -> jint {
+ let vm = match unsafe { JavaVM::from_raw(vm) } {
+ Ok(vm) => vm,
+ Err(_) => return JNI_ERR,
+ };
+ if let Err(error) = install_log_bridge(&vm) {
+ eprintln!("[paimon-vindex] native log bridge disabled, keeping stderr
logging: {error}");
+ }
+ JNI_VERSION_1_8
+}
+
+fn install_log_bridge(vm: &JavaVM) -> Result<(), jni::errors::Error> {
+ // The JNI_OnLoad thread is already attached; FindClass here resolves
+ // through NativeLibraryLoader's classloader (JNI spec), which also loads
+ // NativeLogBridge.
+ let mut env = vm.get_env()?;
+ let found = env.find_class(BRIDGE_CLASS);
+ let class = clear_on_err(&mut env, found)?;
+ let method = env.get_static_method_id(&class, "log",
"(ILjava/lang/String;)V");
+ let log_method = clear_on_err(&mut env, method)?;
+ let class = env.new_global_ref(&class)?;
+ let bridge = LogBridge {
+ vm: unsafe { JavaVM::from_raw(vm.get_java_vm_pointer())? },
+ class,
+ log_method,
+ };
+ let _ = set_log_sink(Box::new(move |level, message| {
+ // Contract: never panic, never leave a pending exception behind.
+ let delivered =
+ catch_unwind(AssertUnwindSafe(|| forward(&bridge, level,
message))).unwrap_or(false);
+ if !delivered {
+ eprintln!("{message}");
+ }
+ }));
+ Ok(())
+}
+
+fn clear_on_err<T>(env: &mut JNIEnv, result: jni::errors::Result<T>) ->
jni::errors::Result<T> {
+ // A failed FindClass/GetStaticMethodID leaves a pending exception which
+ // would make System.load throw; clear it before degrading gracefully.
+ if result.is_err() && env.exception_check().unwrap_or(false) {
+ let _ = env.exception_clear();
+ }
+ result
+}
+
+fn forward(bridge: &LogBridge, level: LogLevel, message: &str) -> bool {
+ // No-op for already-attached threads; daemon-attaches Rayon workers so
+ // they never block DestroyJavaVM. Auto-detach happens at thread exit.
+ let Ok(mut env) = bridge.vm.attach_current_thread_as_daemon() else {
+ return false;
+ };
+ // Daemon-attached worker threads have no Java frame to release locals.
+ env.with_local_frame(4, |env| -> jni::errors::Result<bool> {
+ let jmsg = env.new_string(message)?;
+ let args = [
+ JValue::Int(level as jint).as_jni(),
+ JValue::Object(&jmsg).as_jni(),
+ ];
+ let ok = unsafe {
+ env.call_static_method_unchecked(
+ &bridge.class,
+ bridge.log_method,
+ ReturnType::Primitive(Primitive::Void),
+ &args,
+ )
+ }
+ .is_ok();
+ if !ok || env.exception_check()? {
+ let _ = env.exception_clear();
+ return Ok(false);
+ }
+ Ok(true)
+ })
+ .unwrap_or(false)
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn log_bridge_is_send_and_sync() {
+ fn assert_send_sync<T: Send + Sync>() {}
+ assert_send_sync::<LogBridge>();
+ }
+}
