use super::*;
use crate::segment::chunk_map::{
ChunkMapBuilder, DocLengthsColumn, write_chunk_maps_with_norm_policy,
};
use crate::segment::row_map::RowMap;
use crate::structures::fast_field::{
BLOCK_INDEX_ENTRY_SIZE, FastFieldReader, write_fast_field_toc_and_footer,
};
use crate::structures::{PositionStreamEncoder, SSTableWriter, TermInfo};
use std::io::Write;
fn admit(bytes: usize, budget: usize) -> Result<()> {
if bytes > budget {
return Err(crate::Error::Schema(format!(
"compaction scratch requires {bytes} bytes; remaining budget is {budget}"
)));
}
Ok(())
}
impl SegmentMerger {
pub async fn compact<D: DirectoryWriter>(
&self,
dir: &D,
source: &SegmentReader,
output: SegmentId,
memory_budget: usize,
) -> Result<(SegmentMeta, MergeStats)> {
self.ensure_not_cancelled()?;
let rows = RowMap::from_visibility(
source.num_docs(),
source.num_live_docs(),
source.alive_docs(),
memory_budget / 4,
|| self.ensure_not_cancelled(),
)?;
let budget = memory_budget.saturating_sub(rows.memory_bytes());
for (field, entry) in self.schema.fields() {
if ((entry.indexed && entry.field_type == FieldType::Text)
|| entry.field_type == FieldType::SparseVector)
&& !source.row_stats().contains_key(&field.0)
{
return Err(crate::Error::Schema(format!(
"field '{}' lacks lossless row statistics; rebuild this pre-deletion-format segment before compaction",
entry.name
)));
}
}
let files = SegmentFiles::new(output.0);
let mut stats = MergeStats::default();
let (field_stats, chunk_maps) = self
.compact_text_maps(dir, source, &rows, &files, budget)
.await?;
let posting_budget =
budget.saturating_sub(chunk_maps.values().map(RowMap::memory_bytes).sum::<usize>());
stats.terms_processed = self
.compact_postings(dir, source, &rows, &chunk_maps, &files, posting_budget)
.await?;
drop(chunk_maps);
stats.fast_bytes = self
.compact_columns(dir, &files.fast, source.fast_fields(), &rows, budget)
.await?;
self.compact_columns(dir, &files.row_stats, source.row_stats(), &rows, budget)
.await?;
let mut store_out = OffsetWriter::new(dir.streaming_writer_cold(&files.store).await?);
let mut store = crate::segment::StoreMerger::new(&mut store_out);
store
.append_compacted(source.store(), &rows, self.cancellation.as_deref(), budget)
.await?;
if store.finish()? != rows.len() {
return Err(crate::Error::Corruption(
"compacted store row count mismatch".into(),
));
}
stats.store_bytes = store_out.offset() as usize;
store_out.finish()?;
stats.vectors_bytes = self
.compact_vectors(dir, source, &rows, &files, budget)
.await?;
stats.sparse_bytes = self
.compact_sparse(dir, source, &rows, &files, budget)
.await?;
self.ensure_not_cancelled()?;
let meta = SegmentMeta {
id: output.0,
num_docs: rows.len(),
field_stats,
};
dir.write_durable(&files.meta, &meta.serialize()?).await?;
Ok((meta, stats))
}
async fn compact_columns<D: DirectoryWriter>(
&self,
dir: &D,
path: &std::path::Path,
columns: &FxHashMap<u32, FastFieldReader>,
rows: &RowMap,
budget: usize,
) -> Result<usize> {
use crate::structures::fast_field::BlockIndexEntry;
struct OutputBlock {
source: usize,
range: std::ops::Range<u32>,
header: BlockIndexEntry,
copy: bool,
cached: Option<Vec<u8>>,
}
if columns.is_empty() {
return Ok(0);
}
let mut writer = OffsetWriter::new(dir.streaming_writer_cold(path).await?);
let mut fields: Vec<_> = columns.keys().copied().collect();
fields.sort_unstable();
let mut toc = Vec::new();
for field in fields {
self.ensure_not_cancelled()?;
let reader = &columns[&field];
if reader.num_docs != rows.physical() {
return Err(crate::Error::Corruption(
"compaction column row count mismatch".into(),
));
}
let capacity = reader
.blocks()
.iter()
.map(|block| {
let live =
rows.count(block.cumulative_docs..block.cumulative_docs + block.num_docs);
if live == 0 {
0
} else if live == block.num_docs {
1
} else {
(live as usize).min((block.num_docs as usize).div_ceil(4096))
}
})
.sum::<usize>();
admit(
capacity.saturating_mul(std::mem::size_of::<OutputBlock>()),
budget / 4,
)?;
let mut plan = Vec::with_capacity(capacity);
let mut cached_bytes = 0usize;
for (source, block) in reader.blocks().iter().enumerate() {
self.ensure_not_cancelled()?;
let live =
rows.count(block.cumulative_docs..block.cumulative_docs + block.num_docs);
if live == 0 {
continue;
}
if live == block.num_docs {
plan.push(OutputBlock {
source,
range: 0..block.num_docs,
copy: true,
cached: None,
header: BlockIndexEntry {
num_docs: live,
data_len: block.data.len() as u32,
dict_count: block.dict.as_ref().map_or(0, |dict| dict.len()),
dict_len: block.raw_dict.len() as u32,
},
});
continue;
}
for start in (0..block.num_docs).step_by(4096) {
self.ensure_not_cancelled()?;
let end = start.saturating_add(4096).min(block.num_docs);
if rows.count(block.cumulative_docs + start..block.cumulative_docs + end) == 0 {
continue;
}
let bytes = block.compact_range(
reader.column_type,
reader.multi,
start..end,
|doc| rows.get(doc).is_some(),
budget / 2,
)?;
let header = BlockIndexEntry::read_from(&mut &bytes[4..])?;
let cached = if bytes.capacity() <= (budget / 4).saturating_sub(cached_bytes) {
cached_bytes += bytes.capacity();
Some(bytes)
} else {
None
};
plan.push(OutputBlock {
source,
range: start..end,
header,
copy: false,
cached,
});
}
}
let offset = writer.offset();
writer.write_all(&(plan.len() as u32).to_le_bytes())?;
for block in &plan {
block.header.write_to(&mut writer)?;
}
for block in plan {
self.ensure_not_cancelled()?;
let source = &reader.blocks()[block.source];
if block.copy {
for data in [source.data.as_slice(), source.raw_dict.as_slice()] {
for chunk in data.chunks(4 * 1024 * 1024) {
self.ensure_not_cancelled()?;
writer.write_all(chunk)?;
}
}
} else {
let bytes = match block.cached {
Some(bytes) => bytes,
None => source.compact_range(
reader.column_type,
reader.multi,
block.range,
|doc| rows.get(doc).is_some(),
budget / 2,
)?,
};
let mut header = Vec::with_capacity(BLOCK_INDEX_ENTRY_SIZE);
block.header.write_to(&mut header)?;
if bytes.get(4..4 + BLOCK_INDEX_ENTRY_SIZE) != Some(header.as_slice()) {
return Err(crate::Error::Corruption(
"non-deterministic compacted fast column".into(),
));
}
writer.write_all(&bytes[4 + BLOCK_INDEX_ENTRY_SIZE..])?;
}
}
toc.push(crate::structures::fast_field::FastFieldTocEntry {
field_id: field,
column_type: reader.column_type,
multi: reader.multi,
data_offset: offset,
data_len: writer.offset() - offset,
num_docs: rows.len(),
dict_offset: 0,
dict_count: 0,
});
}
let offset = writer.offset();
write_fast_field_toc_and_footer(&mut writer, offset, &toc)?;
let bytes = writer.offset() as usize;
writer.finish()?;
Ok(bytes)
}
async fn compact_text_maps<D: DirectoryWriter>(
&self,
dir: &D,
source: &SegmentReader,
rows: &RowMap,
files: &SegmentFiles,
budget: usize,
) -> Result<(FxHashMap<u32, FieldStats>, FxHashMap<u32, RowMap>)> {
let mut retained_bytes = 0usize;
let mut statistics = FxHashMap::default();
let mut chunks = Vec::new();
let mut maps = FxHashMap::default();
let mut norms = Vec::new();
for (field, entry) in self.schema.fields() {
if !entry.indexed || entry.field_type != FieldType::Text {
continue;
}
self.ensure_not_cancelled()?;
let exact = &source.row_stats()[&field.0];
let mut stat = FieldStats::default();
for (visited, old) in rows.iter().enumerate() {
if visited.is_multiple_of(4096) {
self.ensure_not_cancelled()?;
}
let value = exact.get_u64(old);
if value > 0 {
stat.doc_count += 1;
stat.total_tokens =
stat.total_tokens.checked_add(value - 1).ok_or_else(|| {
crate::Error::Corruption("compacted token count overflow".into())
})?;
}
}
if source.has_text_mapping(field) {
if entry.chunked {
stat.doc_count = 0;
}
if let Some(source_map) = source.chunk_map(field) {
let map = RowMap::try_new(
source_map.num_chunks(),
source_map.num_chunks(),
|vid| {
if vid.is_multiple_of(4096) {
self.ensure_not_cancelled()?;
}
Ok(rows.get(source_map.doc_id(vid)).is_some())
},
(budget / 2).saturating_sub(retained_bytes),
)?;
retained_bytes = retained_bytes
.saturating_add(map.memory_bytes())
.saturating_add(map.len() as usize * 8);
admit(retained_bytes, budget / 2)?;
let mut builder = ChunkMapBuilder::with_capacity(map.len() as usize);
builder.set_document_units(source_map.is_document_map());
for (visited, old) in map.iter().enumerate() {
if visited.is_multiple_of(4096) {
self.ensure_not_cancelled()?;
}
let (doc, ordinal) = source_map.resolve(old);
builder.push(rows.get(doc).unwrap(), ordinal, source_map.length(old))?;
}
if entry.chunked {
stat.doc_count = map.len();
}
builder.set_total_tokens(stat.total_tokens);
chunks.push((field.0, builder));
maps.insert(field.0, map);
}
} else if let Some(lengths) = source.doc_lengths(field) {
retained_bytes = retained_bytes.saturating_add(rows.len() as usize * 2);
admit(retained_bytes, budget / 2)?;
let mut values = Vec::with_capacity(rows.len() as usize);
for (visited, old) in rows.iter().enumerate() {
if visited.is_multiple_of(4096) {
self.ensure_not_cancelled()?;
}
values.push(lengths.length(old).min(u16::MAX as u32) as u16);
}
norms.push((field.0, values, stat.total_tokens));
}
statistics.insert(field.0, stat);
}
let chunk_refs: Vec<_> = chunks
.iter()
.filter(|(_, map)| !map.is_empty())
.map(|(field, map)| (*field, map))
.collect();
let norm_refs: Vec<_> = norms
.iter()
.map(|(field, lengths, total)| DocLengthsColumn {
field_id: *field,
lengths,
total_tokens: *total,
})
.collect();
if !chunk_refs.is_empty() || !norm_refs.is_empty() {
let mut writer = dir.streaming_writer_cold(&files.chunks).await?;
write_chunk_maps_with_norm_policy(&mut *writer, &chunk_refs, &norm_refs, |field_id| {
source
.doc_lengths(crate::Field(field_id))
.is_some_and(|lengths| lengths.is_quantized())
})?;
writer.finish()?;
}
Ok((statistics, maps))
}
async fn compact_postings<D: DirectoryWriter>(
&self,
dir: &D,
source: &SegmentReader,
rows: &RowMap,
chunks: &FxHashMap<u32, RowMap>,
files: &SegmentFiles,
budget: usize,
) -> Result<usize> {
admit(16 * 1024, budget)?;
let budget = budget - 16 * 1024;
let mut postings = OffsetWriter::new(dir.streaming_writer_cold(&files.postings).await?);
let mut positions = OffsetWriter::new(dir.streaming_writer_cold(&files.positions).await?);
let mut terms_out = OffsetWriter::new(dir.streaming_writer_cold(&files.term_dict).await?);
let mut terms = SSTableWriter::<_, TermInfo>::with_config(
&mut terms_out,
crate::structures::SSTableWriterConfig {
block_size: self.term_dict_block_size,
..crate::structures::SSTableWriterConfig::from_optimization(self.optimization)
},
);
let mut iter = source.term_dict_iter();
let mut count = 0;
while let Some((key, info)) = iter.next().await? {
self.ensure_not_cancelled()?;
let field = crate::Field(u32::from_le_bytes(
key.get(..4)
.ok_or_else(|| crate::Error::Corruption("invalid term field prefix".into()))?
.try_into()
.unwrap(),
));
let map = chunks.get(&field.0).unwrap_or(rows);
let info = if let Some((ids, tfs)) = info.decode_inline() {
let entries: Vec<_> = ids
.into_iter()
.zip(tfs)
.filter_map(|(old, tf)| map.get(old).map(|new| (new, tf)))
.collect();
if entries.is_empty() {
continue;
}
TermInfo::try_inline_iter(entries.len(), entries.into_iter()).ok_or_else(|| {
crate::Error::Corruption("compacted inline posting no longer fits".into())
})?
} else if let Some((offset, len)) = info.external_info() {
use crate::structures::postings::{
PositionRangeSource, PostingBlockSource, PostingStreamWriter,
};
let cancellation = self.cancellation.as_deref();
let input = PostingBlockSource::open(
source.posting_file_range(offset, len)?,
budget / 4,
cancellation,
)
.await?;
if input.doc_count() != info.doc_freq() || input.doc_count() > map.physical() {
return Err(crate::Error::Corruption(
"posting source count disagrees with term or row space".into(),
));
}
let mut source_positions = match info.position_info() {
Some((offset, len)) => Some(
PositionRangeSource::open(
source.position_file_range(offset, len)?,
budget / 4,
cancellation,
)
.await?,
),
None => None,
};
let has_positions = source_positions.is_some();
if input.has_positions() != has_positions {
return Err(crate::Error::Corruption(
"posting and position formats disagree".into(),
));
}
if let Some(positions) = source_positions.as_ref() {
let total = if input.len() == 0 {
0
} else {
input.position_span(input.len() - 1)?.end
};
if positions.total_positions() != total {
return Err(crate::Error::Corruption(
"posting and position counts disagree".into(),
));
}
}
let off = postings.offset();
let pos_off = positions.offset();
let mut output = PostingStreamWriter::new(
&mut postings,
input.len(),
has_positions,
self.posting_codec,
budget / 4,
)?;
if input.is_compact() && self.posting_codec != crate::structures::PostingCodec::Pfor
{
output.enable_compact_headers()?;
}
if input.has_impact_bounds() {
output.enable_impact_bounds()?;
} else if input.has_ratio_bounds() {
output.enable_ratio_bounds()?;
}
let mut position_output = PositionStreamEncoder::with_budget(
&mut positions,
budget / 4,
self.posting_codec,
);
if source_positions
.as_ref()
.is_some_and(PositionRangeSource::is_compact)
{
position_output = position_output.with_compact_directory();
}
let mut docs = Vec::with_capacity(crate::structures::postings::POSTING_BLOCK_SIZE);
let mut tfs = Vec::with_capacity(crate::structures::postings::POSTING_BLOCK_SIZE);
for i in 0..input.len() {
self.ensure_not_cancelled()?;
let (first, last) = input.bounds(i);
if last >= map.physical() {
return Err(crate::Error::Corruption(
"posting address exceeds compaction map".into(),
));
}
let live = map.count(first..last + 1);
if live == 0 {
continue;
}
let block = input.read_block(i).await?;
let span = input.position_span(i)?;
if live == last - first + 1 {
output.append(&block, map.get(first).expect("live interval"))?;
if let Some(positions) = &mut source_positions {
positions
.append_range(&mut position_output, span, cancellation)
.await?;
}
continue;
}
if !block.decode_block_into(0, &mut docs, &mut tfs)
|| docs.first() != Some(&first)
|| docs.last() != Some(&last)
|| !docs.windows(2).all(|pair| pair[0] < pair[1])
|| tfs.contains(&0)
{
return Err(crate::Error::Corruption(
"invalid compacted posting block".into(),
));
}
let mut cursor = span.start;
for (&old, &tf) in docs.iter().zip(&tfs) {
if let Some(new) = map.get(old) {
let length = source.chunk_map(field).map_or_else(
|| source.doc_lengths(field).map_or(1, |norm| norm.length(old)),
|map| map.length(old),
);
output.push(new, tf, length)?;
if let Some(positions) = &mut source_positions {
positions
.append_doc(&mut position_output, cursor, tf, cancellation)
.await?;
}
}
cursor = cursor.checked_add(u64::from(tf)).ok_or_else(|| {
crate::Error::Corruption("position count overflow".into())
})?;
}
if input.has_positions() && cursor != span.end {
return Err(crate::Error::Corruption(
"position cursor disagrees with term frequencies".into(),
));
}
}
if output.doc_count() == 0 {
continue;
}
if output.doc_count() > input.doc_count() {
return Err(crate::Error::Corruption(
"compacted posting count exceeds source".into(),
));
}
let total_positions = output.total_positions();
let (docs, len) = output.finish(cancellation)?;
if has_positions {
self.ensure_not_cancelled()?;
let (total, pos_len) = position_output.finish_cancellable(cancellation)?;
if total != total_positions {
return Err(crate::Error::Corruption(
"compacted position count mismatch".into(),
));
}
TermInfo::external_with_positions(off, len, docs, pos_off, pos_len)
} else {
TermInfo::external(off, len, docs)
}
} else {
return Err(crate::Error::Corruption(
"invalid term posting representation".into(),
));
};
terms.insert(&key, &info)?;
count += 1;
}
terms.finish()?;
terms_out.finish()?;
postings.finish()?;
if positions.offset() > 0 {
positions.finish()?;
} else {
drop(positions);
dir.delete(&files.positions).await?;
}
Ok(count)
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::structures::BlockPostingList;
use crate::structures::fast_field::{FastFieldColumnType, FastFieldWriter};
#[tokio::test]
async fn compaction_preserves_mixed_norm_encodings_per_field() {
use crate::segment::builder::{SegmentBuilder, SegmentBuilderConfig};
use crate::segment::chunk_map::{
read_chunk_maps, write_chunk_maps_with_copied_norms, write_chunk_maps_with_norms,
};
let mut schema = crate::Schema::builder();
let exact = schema.add_text_field("exact", true, false);
let quantized = schema.add_text_field("quantized", true, false);
let schema = Arc::new(schema.build());
let dir = crate::RamDirectory::new();
let id = SegmentId::new();
let mut builder =
SegmentBuilder::new(schema.clone(), SegmentBuilderConfig::default()).unwrap();
for _ in 0..16 {
let mut doc = crate::Document::new();
doc.add_text(exact, "term ".repeat(32));
doc.add_text(quantized, "term ".repeat(32));
builder.add_document(doc).unwrap();
}
builder.build(&dir, id, None).await.unwrap();
let files = SegmentFiles::new(id.0);
let original = read_chunk_maps(
dir.open_read(&files.chunks)
.await
.unwrap()
.read_bytes()
.await
.unwrap(),
)
.unwrap();
let mut bytes = Vec::new();
write_chunk_maps_with_norms(
&mut bytes,
&[],
&[DocLengthsColumn {
field_id: quantized.0,
lengths: &[32; 16],
total_tokens: 512,
}],
true,
)
.unwrap();
let byte_column = read_chunk_maps(crate::directories::OwnedBytes::new(bytes)).unwrap();
let mut mixed = Vec::new();
write_chunk_maps_with_copied_norms(
&mut mixed,
&[],
&[
(exact.0, &original.doc_lengths[&exact.0]),
(quantized.0, &byte_column.doc_lengths[&quantized.0]),
],
)
.unwrap();
dir.write(&files.chunks, &mixed).await.unwrap();
let source = SegmentReader::open(&dir, id, schema.clone(), 4)
.await
.unwrap();
let rows = RowMap::new(16, 16, |doc| doc % 2 == 0, 1024).unwrap();
let output = SegmentFiles::new(SegmentId::new().0);
SegmentMerger::new(schema)
.compact_text_maps(&dir, &source, &rows, &output, 1024 * 1024)
.await
.unwrap();
let columns = read_chunk_maps(
dir.open_read(&output.chunks)
.await
.unwrap()
.read_bytes()
.await
.unwrap(),
)
.unwrap();
for (field, encoded) in [(exact, false), (quantized, true)] {
let lengths = &columns.doc_lengths[&field.0];
assert_eq!(lengths.is_quantized(), encoded);
assert_eq!(lengths.num_docs(), 8);
assert_eq!(lengths.total_tokens(), 256);
assert!((0..8).all(|doc| lengths.length(doc) == 32));
}
}
#[tokio::test]
async fn frequent_terms_compact_with_block_sized_decoding_and_preserve_positions() {
check_frequent_compaction(false).await;
}
#[tokio::test]
async fn compact_layout_survives_dense_and_sparse_row_compaction() {
check_frequent_compaction(true).await;
}
async fn check_frequent_compaction(compact: bool) {
use crate::directories::{Directory, RamDirectory};
use crate::structures::{TERMINATED, TermPositions};
let mut schema = crate::SchemaBuilder::default();
let body = schema.add_text_field("body", true, false);
schema.set_positions(body, crate::dsl::PositionMode::TokenPosition);
let schema = schema.build();
let dir = RamDirectory::new();
let index = crate::Index::create(
dir.clone(),
schema.clone(),
crate::IndexConfig {
compact_text: compact,
quantized_norms: compact,
merge_policy: Box::new(crate::NoMergePolicy),
..Default::default()
},
)
.await
.unwrap();
let mut writer = index.writer();
for _ in 0..8192 {
let mut doc = crate::Document::new();
doc.add_text(body, "common common common common");
writer.add_document(doc).unwrap();
}
writer.commit().await.unwrap();
let reader = index.reader().await.unwrap();
let searcher = reader.searcher().await.unwrap();
for sparse in [false, true] {
let rows = RowMap::new(
8192,
8192,
|doc| {
if sparse {
doc % 129 == 0
} else {
doc >= 256 && doc % 257 != 0
}
},
4096,
)
.unwrap();
let merger = SegmentMerger::new(Arc::new(schema.clone()));
let files = SegmentFiles::new(SegmentId::new().0);
merger
.compact_postings(
&dir,
&searcher.segment_readers()[0],
&rows,
&FxHashMap::default(),
&files,
64 * 1024,
)
.await
.unwrap();
let posting_bytes = dir
.open_read(&files.postings)
.await
.unwrap()
.read_bytes()
.await
.unwrap();
if compact {
let flags = u32::from_le_bytes(
posting_bytes[posting_bytes.len() - 12..posting_bytes.len() - 8]
.try_into()
.unwrap(),
);
assert_ne!(flags & 64, 0, "compaction retains separated headers");
}
let postings = BlockPostingList::deserialize(posting_bytes.as_slice()).unwrap();
assert_eq!(postings.doc_count(), rows.len());
if sparse {
assert_eq!(
postings.num_blocks(),
1,
"sparse survivors should share a rebuilt block"
);
}
let position_bytes = dir
.open_read(&files.positions)
.await
.unwrap()
.read_bytes()
.await
.unwrap();
if compact {
assert_eq!(&position_bytes[position_bytes.len() - 4..], b"POS6");
}
let positions = TermPositions::open(position_bytes).unwrap();
let mut cursor = postings.iterator();
let mut scratch = Vec::new();
let mut values = Vec::new();
for doc in 0..rows.len() {
assert_eq!(cursor.doc(), doc);
assert_eq!(cursor.term_freq(), 4);
assert!(positions.positions_into(
cursor.position_cursor(),
4,
&mut scratch,
&mut values
));
assert_eq!(values, [0, 1, 2, 3]);
cursor.advance();
}
assert_eq!(cursor.doc(), TERMINATED);
}
}
#[tokio::test]
async fn intact_fast_blocks_keep_encoded_bytes_when_earlier_rows_are_deleted() {
let n = 8193;
let mut column = FastFieldWriter::new_numeric(FastFieldColumnType::U64);
for i in 0..n {
column.add_u64(i, u64::from(i) * 17);
}
let mut encoded = Vec::new();
let (mut toc, _) = column.serialize(&mut encoded, 0).unwrap();
let header = &encoded[4..4 + BLOCK_INDEX_ENTRY_SIZE];
let payload = &encoded[4 + BLOCK_INDEX_ENTRY_SIZE..];
let mut stacked = 2u32.to_le_bytes().to_vec();
stacked.extend_from_slice(header);
stacked.extend_from_slice(header);
stacked.extend_from_slice(payload);
stacked.extend_from_slice(payload);
toc.num_docs = 2 * n;
toc.data_len = stacked.len() as u64;
let source =
FastFieldReader::open(&crate::directories::OwnedBytes::new(stacked), &toc).unwrap();
let columns = FxHashMap::from_iter([(0, source)]);
let rows = RowMap::new(2 * n, n, |doc| doc >= n, 1024 * 1024).unwrap();
let dir = crate::directories::RamDirectory::new();
let merger = SegmentMerger::new(Arc::new(crate::SchemaBuilder::default().build()));
let path = std::path::Path::new("copied.fast");
merger
.compact_columns(&dir, path, &columns, &rows, 4 * 1024 * 1024)
.await
.unwrap();
let bytes = dir
.open_read(path)
.await
.unwrap()
.read_bytes()
.await
.unwrap();
let (offset, count) =
crate::structures::fast_field::read_fast_field_footer(bytes.as_slice()).unwrap();
let toc =
crate::structures::fast_field::read_fast_field_toc(bytes.as_slice(), offset, count)
.unwrap();
assert_eq!(toc[0].data_len as usize, encoded.len());
assert_eq!(&bytes.as_slice()[..encoded.len()], encoded.as_slice());
}
#[tokio::test]
async fn cached_and_reencoded_column_chunks_have_identical_bytes_and_missing_rows() {
let count = 2 * (32 * 4096 + 1);
let mut column = FastFieldWriter::new_numeric(FastFieldColumnType::U64);
for doc in 0..count {
if doc % 97 != 0 {
column.add_u64(
doc,
u64::from(doc)
.wrapping_mul(0x9e3779b97f4a7c15)
.rotate_left(23),
);
}
}
column.pad_to(count);
let mut encoded = Vec::new();
let (toc, _) = column.serialize(&mut encoded, 0).unwrap();
let source =
FastFieldReader::open(&crate::directories::OwnedBytes::new(encoded), &toc).unwrap();
let columns = FxHashMap::from_iter([(0, source)]);
let rows = RowMap::new(count, count / 2, |doc| doc % 2 == 1, 8 * 1024 * 1024).unwrap();
let dir = crate::directories::RamDirectory::new();
let merger = SegmentMerger::new(Arc::new(crate::SchemaBuilder::default().build()));
let mut outputs = Vec::new();
for (name, budget) in [
("partial.fast", 4 * 1024 * 1024),
("cached.fast", 16 * 1024 * 1024),
] {
let path = std::path::Path::new(name);
merger
.compact_columns(&dir, path, &columns, &rows, budget)
.await
.unwrap();
outputs.push(
dir.open_read(path)
.await
.unwrap()
.read_bytes()
.await
.unwrap(),
);
}
assert!(
outputs[0].len() > 1024 * 1024,
"fixture no longer exercises cache overflow"
);
assert_eq!(outputs[0].as_slice(), outputs[1].as_slice());
let (offset, count) =
crate::structures::fast_field::read_fast_field_footer(outputs[0].as_slice()).unwrap();
let toc = crate::structures::fast_field::read_fast_field_toc(
outputs[0].as_slice(),
offset,
count,
)
.unwrap();
let actual = FastFieldReader::open(&outputs[0], &toc[0]).unwrap();
assert_eq!(actual.num_docs, rows.len());
for (new, old) in rows.iter().enumerate() {
assert_eq!(actual.get_u64(new as u32), columns[&0].get_u64(old));
assert_eq!(actual.has_value(new as u32), columns[&0].has_value(old));
}
}
}