From 7717a96baf4248a3e80146957b3067fc35b73b9e Mon Sep 17 00:00:00 2001 From: discord9 <55937128+discord9@users.noreply.github.com> Date: Tue, 29 Sep 2026 17:36:49 +0800 Subject: [PATCH] perf(index): decode inverted postings as they are unioned Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> --- .../search/fst_values_mapper.rs | 378 ++++++++++++++---- .../search/index_apply/predicates_apply.rs | 66 +-- 2 files changed, 325 insertions(+), 119 deletions(-) diff --git a/src/index/src/inverted_index/search/fst_values_mapper.rs b/src/index/src/inverted_index/search/fst_values_mapper.rs index 38df713c8d1..18a7bfdc24d 100644 --- a/src/index/src/inverted_index/search/fst_values_mapper.rs +++ b/src/index/src/inverted_index/search/fst_values_mapper.rs @@ -12,10 +12,13 @@ // See the License for the specific language governing permissions and // limitations under the License. +use std::io; + use greptime_proto::v1::index::{BitmapType, InvertedIndexMeta}; +use snafu::{IntoError, ResultExt}; use crate::bitmap::Bitmap; -use crate::inverted_index::error::Result; +use crate::inverted_index::error::{CommonIoSnafu, DecodeBitmapSnafu, Result}; use crate::inverted_index::format::reader::{InvertedIndexReadMetrics, InvertedIndexReader}; /// `ParallelFstValuesMapper` enables parallel mapping of multiple FST value groups to their @@ -33,6 +36,9 @@ impl<'a> ParallelFstValuesMapper<'a> { Self { reader } } + /// Maps each group of FST values to the union of the bitmaps its values point to, preserving the + /// group order. All bitmaps are fetched with a single batched read and decoded one at a time; + /// any decode failure returns an error instead of a partial result. pub async fn map_values_vec( &mut self, value_and_meta_vec: &[(Vec, &InvertedIndexMeta)], @@ -44,6 +50,7 @@ impl<'a> ParallelFstValuesMapper<'a> { .collect::>(); let len = groups.iter().sum::(); let mut fetch_ranges = Vec::with_capacity(len); + let mut bitmap_types = Vec::with_capacity(len); for (values, meta) in value_and_meta_vec { for value in values { @@ -51,12 +58,10 @@ impl<'a> ParallelFstValuesMapper<'a> { // bitmap offset and the lower 32 bits represent its size. This mapper uses these // combined offset-size pairs to fetch and union multiple bitmaps into a single `BitVec`. let [relative_offset, size] = bytemuck::cast::(*value); - let range = meta.base_offset + relative_offset as u64 - ..meta.base_offset + relative_offset as u64 + size as u64; - fetch_ranges.push(( - range, - BitmapType::try_from(meta.bitmap_type).unwrap_or(BitmapType::BitVec), - )); + let start = meta.base_offset + relative_offset as u64; + fetch_ranges.push(start..start + size as u64); + bitmap_types + .push(BitmapType::try_from(meta.bitmap_type).unwrap_or(BitmapType::BitVec)); } } @@ -65,14 +70,31 @@ impl<'a> ParallelFstValuesMapper<'a> { } common_telemetry::debug!("fetch ranges: {:?}", fetch_ranges); - let mut bitmaps = self.reader.bitmap_deque(&fetch_ranges, metrics).await?; + let bytes_vec = self.reader.read_vec(&fetch_ranges, metrics).await?; + // A conforming reader returns exactly one buffer per range. Bail out instead of silently + // unioning a group from fewer bitmaps, which would drop matching rows. + if bytes_vec.len() != fetch_ranges.len() { + let error = io::Error::new( + io::ErrorKind::InvalidData, + format!( + "Expected {} bitmap buffers but got {}", + fetch_ranges.len(), + bytes_vec.len() + ), + ); + return Err(CommonIoSnafu.into_error(error)); + } + + // `bitmap_types` is built alongside `fetch_ranges`, so every buffer keeps its bitmap type. + let mut fetched = bytes_vec.into_iter().zip(bitmap_types); let mut output = Vec::with_capacity(groups.len()); for counter in groups { let mut bitmap = Bitmap::new_roaring(); - for _ in 0..counter { - let bm = bitmaps.pop_front().unwrap(); - bitmap.union(bm); + for (bytes, bitmap_type) in fetched.by_ref().take(counter) { + let decoded = + Bitmap::deserialize_from(&bytes, bitmap_type).context(DecodeBitmapSnafu)?; + bitmap.union(decoded); } output.push(bitmap); @@ -85,117 +107,299 @@ impl<'a> ParallelFstValuesMapper<'a> { #[cfg(test)] mod tests { use std::collections::VecDeque; + use std::io; + use std::ops::Range; + + use bytes::Bytes; + use snafu::Location; use super::*; + use crate::inverted_index::error::Error; use crate::inverted_index::format::reader::MockInvertedIndexReader; fn value(offset: u32, size: u32) -> u64 { bytemuck::cast::<[u32; 2], u64>([offset, size]) } - #[tokio::test] - async fn test_map_values_vec() { + fn meta(bitmap_type: BitmapType) -> InvertedIndexMeta { + InvertedIndexMeta { + bitmap_type: bitmap_type.into(), + ..Default::default() + } + } + + fn roaring(lsb0_bytes: &[u8]) -> Bitmap { + Bitmap::from_lsb0_bytes(lsb0_bytes, BitmapType::Roaring) + } + + fn serialized(bitmap: &Bitmap, bitmap_type: BitmapType) -> Bytes { + let mut buf = Vec::new(); + bitmap.serialize_into(bitmap_type, &mut buf).unwrap(); + Bytes::from(buf) + } + + /// Returns a mock reader that answers exactly one `read_vec` call with the serialized + /// `responses`, asserting that the requested ranges are exactly `expected_ranges` in order. + fn mock_reader_returning( + expected_ranges: Vec>, + responses: Vec<(Bitmap, BitmapType)>, + ) -> MockInvertedIndexReader { let mut mock_reader = MockInvertedIndexReader::new(); mock_reader - .expect_bitmap_deque() - .returning(|ranges, _metrics| { - let mut output = VecDeque::new(); - for (range, bitmap_type) in ranges { - let offset = range.start; - let size = range.end - range.start; - match (offset, size, bitmap_type) { - (1, 1, BitmapType::Roaring) => { - output.push_back(Bitmap::from_lsb0_bytes(&[0b10101010], *bitmap_type)) - } - (2, 1, BitmapType::Roaring) => { - output.push_back(Bitmap::from_lsb0_bytes(&[0b01010101], *bitmap_type)) - } - _ => unreachable!(), - } - } - Ok(output) + .expect_read_vec() + .times(1) + .returning(move |ranges, _metrics| { + assert_eq!(ranges, expected_ranges.as_slice()); + Ok(responses + .iter() + .map(|(bitmap, bitmap_type)| serialized(bitmap, *bitmap_type)) + .collect()) }); + mock_reader + } - let meta = InvertedIndexMeta { - bitmap_type: BitmapType::Roaring.into(), - ..Default::default() - }; + #[tokio::test] + async fn test_map_values_vec_with_empty_values() { + let roaring_meta = meta(BitmapType::Roaring); + let bitvec_meta = meta(BitmapType::BitVec); + // No `read_vec` expectation: no read is issued when every group is empty. + let mut mock_reader = MockInvertedIndexReader::new(); let mut values_mapper = ParallelFstValuesMapper::new(&mut mock_reader); - let result = values_mapper - .map_values_vec(&[(vec![], &meta)], None) - .await - .unwrap(); - assert_eq!(result[0].count_ones(), 0); + let output = values_mapper.map_values_vec(&[], None).await.unwrap(); + assert_eq!(output, vec![Bitmap::new_bitvec()]); - let result = values_mapper - .map_values_vec(&[(vec![value(1, 1)], &meta)], None) + let output = values_mapper + .map_values_vec(&[(vec![], &roaring_meta)], None) .await .unwrap(); - assert_eq!( - result[0], - Bitmap::from_lsb0_bytes(&[0b10101010], BitmapType::Roaring) + assert_eq!(output, vec![Bitmap::new_bitvec()]); + + // All groups empty keeps the existing single empty `BitVec` shape. + let output = values_mapper + .map_values_vec(&[(vec![], &roaring_meta), (vec![], &bitvec_meta)], None) + .await + .unwrap(); + assert_eq!(output, vec![Bitmap::new_bitvec()]); + } + + #[tokio::test] + async fn test_map_values_vec_single_group_in_value_order() { + let roaring_meta = meta(BitmapType::Roaring); + // Values are fetched in the order they are given, duplicates included. + let mut mock_reader = mock_reader_returning( + vec![2..3, 1..2, 1..2], + vec![ + (roaring(&[0b01010101]), BitmapType::Roaring), + (roaring(&[0b10101010]), BitmapType::Roaring), + (roaring(&[0b10101010]), BitmapType::Roaring), + ], ); + let mut values_mapper = ParallelFstValuesMapper::new(&mut mock_reader); - let result = values_mapper - .map_values_vec(&[(vec![value(2, 1)], &meta)], None) - .await - .unwrap(); - assert_eq!( - result[0], - Bitmap::from_lsb0_bytes(&[0b01010101], BitmapType::Roaring) - ); - - let result = values_mapper - .map_values_vec(&[(vec![value(1, 1), value(2, 1)], &meta)], None) - .await - .unwrap(); - assert_eq!( - result[0], - Bitmap::from_lsb0_bytes(&[0b11111111], BitmapType::Roaring) - ); - - let result = values_mapper - .map_values_vec(&[(vec![value(2, 1), value(1, 1)], &meta)], None) - .await - .unwrap(); - assert_eq!( - result[0], - Bitmap::from_lsb0_bytes(&[0b11111111], BitmapType::Roaring) - ); - - let result = values_mapper + let output = values_mapper .map_values_vec( - &[(vec![value(2, 1)], &meta), (vec![value(1, 1)], &meta)], + &[(vec![value(2, 1), value(1, 1), value(1, 1)], &roaring_meta)], None, ) .await .unwrap(); - assert_eq!( - result[0], - Bitmap::from_lsb0_bytes(&[0b01010101], BitmapType::Roaring) + + assert_eq!(output, vec![roaring(&[0b11111111])]); + } + + #[tokio::test] + async fn test_map_values_vec_multiple_groups_with_empty_middle() { + let roaring_meta = meta(BitmapType::Roaring); + let mut mock_reader = mock_reader_returning( + vec![2..3, 1..2, 4..5], + vec![ + (roaring(&[0b01010101]), BitmapType::Roaring), + (roaring(&[0b10101010]), BitmapType::Roaring), + (roaring(&[0b00001111]), BitmapType::Roaring), + ], ); - assert_eq!( - result[1], - Bitmap::from_lsb0_bytes(&[0b10101010], BitmapType::Roaring) - ); - let result = values_mapper + let mut values_mapper = ParallelFstValuesMapper::new(&mut mock_reader); + + let output = values_mapper .map_values_vec( &[ - (vec![value(2, 1), value(1, 1)], &meta), - (vec![value(1, 1)], &meta), + (vec![value(2, 1), value(1, 1)], &roaring_meta), + (vec![], &roaring_meta), + (vec![value(4, 1)], &roaring_meta), ], None, ) .await .unwrap(); + assert_eq!( - result[0], - Bitmap::from_lsb0_bytes(&[0b11111111], BitmapType::Roaring) + output, + vec![ + roaring(&[0b11111111]), + Bitmap::new_roaring(), + roaring(&[0b00001111]), + ] ); + } + + #[tokio::test] + async fn test_map_values_vec_duplicate_values_and_both_encodings() { + let roaring_meta = meta(BitmapType::Roaring); + let bitvec_meta = meta(BitmapType::BitVec); + let mut mock_reader = mock_reader_returning( + vec![1..2, 1..2, 2..3], + vec![ + (roaring(&[0b10101010]), BitmapType::Roaring), + (roaring(&[0b10101010]), BitmapType::Roaring), + ( + Bitmap::from_lsb0_bytes(&[0b01010101], BitmapType::BitVec), + BitmapType::BitVec, + ), + ], + ); + let mut values_mapper = ParallelFstValuesMapper::new(&mut mock_reader); + + let output = values_mapper + .map_values_vec( + &[ + (vec![value(1, 1), value(1, 1)], &roaring_meta), + (vec![value(2, 1)], &bitvec_meta), + ], + None, + ) + .await + .unwrap(); + assert_eq!( - result[1], - Bitmap::from_lsb0_bytes(&[0b10101010], BitmapType::Roaring) + output, + vec![ + roaring(&[0b10101010]), + Bitmap::from_lsb0_bytes(&[0b01010101], BitmapType::BitVec), + ] + ); + } + + #[tokio::test] + async fn test_map_values_vec_late_malformed_bitmap_returns_error() { + let roaring_meta = meta(BitmapType::Roaring); + let mut mock_reader = MockInvertedIndexReader::new(); + mock_reader + .expect_read_vec() + .times(1) + .returning(|ranges, _metrics| { + assert_eq!(ranges, [1..2, 2..3].as_slice()); + Ok(vec![ + serialized(&roaring(&[0b10101010]), BitmapType::Roaring), + Bytes::from_static(b"not a roaring bitmap"), + ]) + }); + let mut values_mapper = ParallelFstValuesMapper::new(&mut mock_reader); + + let output = values_mapper + .map_values_vec(&[(vec![value(1, 1), value(2, 1)], &roaring_meta)], None) + .await; + + assert!(matches!(output, Err(Error::DecodeBitmap { .. }))); + } + + #[tokio::test] + async fn test_map_values_vec_reader_failure_returns_error() { + let roaring_meta = meta(BitmapType::Roaring); + let mut mock_reader = MockInvertedIndexReader::new(); + mock_reader + .expect_read_vec() + .times(1) + .returning(|_ranges, _metrics| { + Err(Error::Read { + error: io::Error::other("read failed"), + location: Location::default(), + }) + }); + let mut values_mapper = ParallelFstValuesMapper::new(&mut mock_reader); + + let output = values_mapper + .map_values_vec(&[(vec![value(1, 1)], &roaring_meta)], None) + .await; + + assert!(matches!(output, Err(Error::Read { .. }))); + } + + #[tokio::test] + async fn test_map_values_vec_short_read_response_returns_error() { + let roaring_meta = meta(BitmapType::Roaring); + let mut mock_reader = MockInvertedIndexReader::new(); + mock_reader + .expect_read_vec() + .times(1) + .returning(|ranges, _metrics| { + assert_eq!(ranges, [1..2, 2..3].as_slice()); + // Fewer buffers than requested must not silently produce a partial group. + Ok(vec![serialized( + &roaring(&[0b10101010]), + BitmapType::Roaring, + )]) + }); + let mut values_mapper = ParallelFstValuesMapper::new(&mut mock_reader); + + let output = values_mapper + .map_values_vec(&[(vec![value(1, 1), value(2, 1)], &roaring_meta)], None) + .await; + + assert!(matches!(output, Err(Error::CommonIo { .. }))); + } + + #[tokio::test] + async fn test_map_values_vec_matches_eager_decode_reference() { + let roaring_meta = meta(BitmapType::Roaring); + let bitvec_meta = meta(BitmapType::BitVec); + let responses = vec![ + (roaring(&[0b10101010]), BitmapType::Roaring), + (roaring(&[0b01010101]), BitmapType::Roaring), + ( + Bitmap::from_lsb0_bytes(&[0b00001111], BitmapType::BitVec), + BitmapType::BitVec, + ), + ]; + let mut mock_reader = mock_reader_returning(vec![1..2, 2..3, 3..4], responses.clone()); + let mut values_mapper = ParallelFstValuesMapper::new(&mut mock_reader); + + let output = values_mapper + .map_values_vec( + &[ + (vec![value(1, 1), value(2, 1)], &roaring_meta), + (vec![], &roaring_meta), + (vec![value(3, 1)], &bitvec_meta), + ], + None, + ) + .await + .unwrap(); + + // The previous eager behavior: decode every fetched bitmap first, then pop and union + // them into each group in order. + let mut decoded = VecDeque::new(); + for (bitmap, bitmap_type) in &responses { + let bytes = serialized(bitmap, *bitmap_type); + decoded.push_back(Bitmap::deserialize_from(&bytes, *bitmap_type).unwrap()); + } + let mut eager = Vec::new(); + for counter in [2usize, 0, 1] { + let mut bitmap = Bitmap::new_roaring(); + for _ in 0..counter { + bitmap.union(decoded.pop_front().unwrap()); + } + eager.push(bitmap); + } + + assert_eq!(output, eager); + assert_eq!( + output, + vec![ + roaring(&[0b11111111]), + Bitmap::new_roaring(), + Bitmap::from_lsb0_bytes(&[0b00001111], BitmapType::BitVec), + ] ); } } diff --git a/src/index/src/inverted_index/search/index_apply/predicates_apply.rs b/src/index/src/inverted_index/search/index_apply/predicates_apply.rs index 5fcc962ca7d..599020e660e 100644 --- a/src/index/src/inverted_index/search/index_apply/predicates_apply.rs +++ b/src/index/src/inverted_index/search/index_apply/predicates_apply.rs @@ -167,9 +167,9 @@ impl TryFrom)>> for PredicatesIndexApplier { #[cfg(test)] mod tests { - use std::collections::VecDeque; use std::sync::Arc; + use bytes::Bytes; use greptime_proto::v1::index::{BitmapType, InvertedIndexMeta}; use super::*; @@ -213,6 +213,13 @@ mod tests { bytemuck::cast::<_, u64>([offset, size]) } + /// Serializes `bitmap` so it can be returned as a raw `read_vec` response. + fn serialized(bitmap: &Bitmap, bitmap_type: BitmapType) -> Bytes { + let mut buf = Vec::new(); + bitmap.serialize_into(bitmap_type, &mut buf).unwrap(); + Bytes::from(buf) + } + #[tokio::test] async fn test_index_applier_apply_get_key() { // An index applier that point-gets "tag-0_value-0" on tag "tag-0" @@ -231,19 +238,14 @@ mod tests { ]) }); - mock_reader - .expect_bitmap_deque() - .returning(|arg, _metrics| { - assert_eq!(arg.len(), 1); - let range = &arg[0].0; - let bitmap_type = arg[0].1; - assert_eq!(*range, 2..3); - assert_eq!(bitmap_type, BitmapType::Roaring); - Ok(VecDeque::from([Bitmap::from_lsb0_bytes( - &[0b10101010], - bitmap_type, - )])) - }); + mock_reader.expect_read_vec().returning(|ranges, _metrics| { + assert_eq!(ranges.len(), 1); + assert_eq!(ranges[0], 2..3); + Ok(vec![serialized( + &Bitmap::from_lsb0_bytes(&[0b10101010], BitmapType::Roaring), + BitmapType::Roaring, + )]) + }); let output = applier .apply(SearchContext::default(), &mut mock_reader, None) .await @@ -298,26 +300,26 @@ mod tests { } Ok(output) }); - mock_reader - .expect_bitmap_deque() - .returning(|ranges, _metrics| { - let mut output = VecDeque::new(); - for (range, bitmap_type) in ranges { - let offset = range.start; - let size = range.end - range.start; - match (offset, size, bitmap_type) { - (1, 1, BitmapType::Roaring) => { - output.push_back(Bitmap::from_lsb0_bytes(&[0b10101010], *bitmap_type)) - } - (2, 1, BitmapType::Roaring) => { - output.push_back(Bitmap::from_lsb0_bytes(&[0b11011011], *bitmap_type)) - } - _ => unreachable!(), - } + mock_reader.expect_read_vec().returning(|ranges, _metrics| { + let mut output = Vec::with_capacity(ranges.len()); + for range in ranges { + let offset = range.start; + let size = range.end - range.start; + match (offset, size) { + (1, 1) => output.push(serialized( + &Bitmap::from_lsb0_bytes(&[0b10101010], BitmapType::Roaring), + BitmapType::Roaring, + )), + (2, 1) => output.push(serialized( + &Bitmap::from_lsb0_bytes(&[0b11011011], BitmapType::Roaring), + BitmapType::Roaring, + )), + _ => unreachable!(), } + } - Ok(output) - }); + Ok(output) + }); let output = applier .apply(SearchContext::default(), &mut mock_reader, None)