rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-10-12 15:02:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

perf(index): decode inverted postings as they are unioned

Browse Source 
Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
This commit is contained in:
discord9
2026-09-29 18:16:43 +08:00
parent fa5f44ac60
commit 7717a96baf
2 changed files with 325 additions and 119 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/index/src/inverted_index/search/fst_values_mapper.rs
+291 -87
View File
@@ -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<u64>, &InvertedIndexMeta)],
@@ -44,6 +50,7 @@ impl<'a> ParallelFstValuesMapper<'a> {
.collect::<Vec<_>>();
let len = groups.iter().sum::<usize>();
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::<u64, [u32; 2]>(*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<Range<u64>>,
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),
]
);
}
}
src/index/src/inverted_index/search/index_apply/predicates_apply.rs
+34 -32
View File
@@ -167,9 +167,9 @@ impl TryFrom<Vec<(String, Vec<Predicate>)>> 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)