Compact multiple segments into a single merged segment. Reads all source segments, skips deleted rows from each, and writes a single merged output segment. This reduces segment count and reclaims space from deleted rows across all sources. When `kek` is `Some`, the merged output segment is AES-256-GCM encrypted. Input segments must be pre-decrypted plaintext. `governor` is optional: when `Some`
(
segments: &[(&[u8], &DeleteBitmap)],
schema: &ColumnarSchema,
profile_tag: u8,
governor: Option<&Arc<MemoryGovernor>>,
kek: Option<&nodedb_wal::crypto::WalEncryptionKey>,
)
| 29 | /// tracked against the `Columnar` engine budget. Pass `None` in embedded |
| 30 | /// (Lite) deployments where no governor is configured. |
| 31 | pub fn compact_segments( |
| 32 | segments: &[(&[u8], &DeleteBitmap)], |
| 33 | schema: &ColumnarSchema, |
| 34 | profile_tag: u8, |
| 35 | governor: Option<&Arc<MemoryGovernor>>, |
| 36 | kek: Option<&nodedb_wal::crypto::WalEncryptionKey>, |
| 37 | ) -> Result<CompactionResult, ColumnarError> { |
| 38 | let mut memtable = ColumnarMemtable::new(schema); |
| 39 | let mut total_removed = 0usize; |
| 40 | let col_len = schema.columns.len(); |
| 41 | let _row_guard = governor |
| 42 | .map(|g| { |
| 43 | g.reserve( |
| 44 | EngineId::Columnar, |
| 45 | col_len * std::mem::size_of::<usize>() * 3, |
| 46 | ) |
| 47 | }) |
| 48 | .transpose()?; |
| 49 | let mut row_values = Vec::with_capacity(col_len); |
| 50 | |
| 51 | for &(segment_data, deletes) in segments { |
| 52 | let reader = SegmentReader::open(segment_data)?; |
| 53 | let total_rows = reader.row_count() as usize; |
| 54 | |
| 55 | let col_count = reader.column_count(); |
| 56 | let _cols_guard = governor |
| 57 | .map(|g| { |
| 58 | g.reserve( |
| 59 | EngineId::Columnar, |
| 60 | col_count * std::mem::size_of::<usize>() * 3, |
| 61 | ) |
| 62 | }) |
| 63 | .transpose()?; |
| 64 | let mut decoded_cols = Vec::with_capacity(col_count); |
| 65 | for i in 0..col_count { |
| 66 | decoded_cols.push(reader.read_column(i)?); |
| 67 | } |
| 68 | |
| 69 | for row_idx in 0..total_rows { |
| 70 | if deletes.is_deleted(row_idx as u32) { |
| 71 | total_removed += 1; |
| 72 | continue; |
| 73 | } |
| 74 | |
| 75 | row_values.clear(); |
| 76 | for (col_idx, decoded) in decoded_cols.iter().enumerate() { |
| 77 | let col = &schema.columns[col_idx]; |
| 78 | let value = extract_row_value(decoded, row_idx, &col.column_type, &col.name)?; |
| 79 | row_values.push(value); |
| 80 | } |
| 81 | |
| 82 | memtable.append_row(&row_values)?; |
| 83 | } |
| 84 | } |
| 85 | |
| 86 | let live_rows = memtable.row_count(); |
| 87 | if live_rows == 0 { |
| 88 | return Ok(CompactionResult { |