From 057468e27cc5e71cdd7333285848c09b750936a9 Mon Sep 17 00:00:00 2001 From: Heikki Linnakangas Date: Thu, 12 May 2022 02:04:11 +0300 Subject: [PATCH] More work on compression Move the responsibility for training the dictionary into DeltaLayerWriter, so that the callers don't need to know about it. Add comments. If the compressed version of a blob would be larger than the original, store it uncompressed. --- pageserver/src/config.rs | 1 + pageserver/src/layered_repository.rs | 161 +++----- .../src/layered_repository/delta_layer.rs | 382 +++++++++++++----- .../src/layered_repository/image_layer.rs | 123 +++++- .../src/layered_repository/inmemory_layer.rs | 34 +- 5 files changed, 463 insertions(+), 238 deletions(-) diff --git a/pageserver/src/config.rs b/pageserver/src/config.rs index 9ee2ca468e..caf2d0a9b3 100644 --- a/pageserver/src/config.rs +++ b/pageserver/src/config.rs @@ -23,6 +23,7 @@ use crate::tenant_config::{TenantConf, TenantConfOpt}; pub const ZSTD_MAX_SAMPLES: usize = 1024; pub const ZSTD_MIN_SAMPLES: usize = 8; // magic requirement of zstd +pub const ZSTD_MAX_SAMPLE_BYTES: usize = 10 * 1024 * 1024; // max memory size for holding samples pub const ZSTD_MAX_DICTIONARY_SIZE: usize = 8 * 1024; pub const ZSTD_COMPRESSION_LEVEL: i32 = 0; // default compression level diff --git a/pageserver/src/layered_repository.rs b/pageserver/src/layered_repository.rs index facc484c0a..01c2b961eb 100644 --- a/pageserver/src/layered_repository.rs +++ b/pageserver/src/layered_repository.rs @@ -17,7 +17,6 @@ use fail::fail_point; use itertools::Itertools; use lazy_static::lazy_static; use tracing::*; -use utils::bin_ser::BeSer; use std::cmp::{max, min, Ordering}; use std::collections::hash_map::Entry; @@ -33,7 +32,6 @@ use std::sync::{Arc, Mutex, MutexGuard, RwLock, RwLockReadGuard, TryLockError}; use std::time::{Duration, Instant, SystemTime}; use self::metadata::{metadata_path, TimelineMetadata, METADATA_FILE_NAME}; -use crate::config; use crate::config::PageServerConf; use crate::keyspace::KeySpace; use crate::storage_sync::index::RemoteIndex; @@ -1963,109 +1961,78 @@ impl LayeredTimeline { .map(|l| l.get_lsn_range()) .reduce(|a, b| min(a.start, b.start)..max(a.end, b.end)) .unwrap(); + + let all_values_iter = level0_deltas.iter().map(|l| l.iter()).kmerge_by(|a, b| { + if let Ok((a_key, a_lsn, _)) = a { + if let Ok((b_key, b_lsn, _)) = b { + match a_key.cmp(b_key) { + Ordering::Less => true, + Ordering::Equal => a_lsn <= b_lsn, + Ordering::Greater => false, + } + } else { + false + } + } else { + true + } + }); + + // Merge the contents of all the input delta layers into a new set + // of delta layers, based on the current partitioning. + // + // TODO: this actually divides the layers into fixed-size chunks, not + // based on the partitioning. + // + // TODO: we should also opportunistically materialize and + // garbage collect what we can. let mut new_layers = Vec::new(); - { - let mut all_values_iter = level0_deltas.iter().map(|l| l.iter()).kmerge_by(|a, b| { - if let Ok((a_key, a_lsn, _)) = a { - if let Ok((b_key, b_lsn, _)) = b { - match a_key.cmp(b_key) { - Ordering::Less => true, - Ordering::Equal => a_lsn <= b_lsn, - Ordering::Greater => false, - } - } else { - false - } - } else { - true - } - }); + let mut prev_key: Option = None; + let mut writer: Option = None; + for x in all_values_iter { + let (key, lsn, value) = x?; - // Merge the contents of all the input delta layers into a new set - // of delta layers, based on the current partitioning. - // - // TODO: this actually divides the layers into fixed-size chunks, not - // based on the partitioning. - // - // TODO: we should also opportunistically materialize and - // garbage collect what we can. - let mut prev_key: Option = None; - let mut writer: Option = None; - - while let Some(x) = all_values_iter.next() { - let (key, lsn, value) = x?; - - if let Some(prev_key) = prev_key { - if key != prev_key && writer.is_some() { - let size = writer.as_mut().unwrap().size(); - if size > target_file_size { - new_layers.push(writer.take().unwrap().finish(prev_key.next())?); - writer = None; - } + if let Some(prev_key) = prev_key { + if key != prev_key && writer.is_some() { + let size = writer.as_mut().unwrap().size(); + if size > target_file_size { + new_layers.push(writer.take().unwrap().finish(prev_key.next())?); + writer = None; } } - - if writer.is_none() { - let mut samples: Vec> = Vec::with_capacity(config::ZSTD_MAX_SAMPLES); - let mut prefetched = Vec::with_capacity(config::ZSTD_MAX_SAMPLES); - samples.push(Value::ser(&value)?); - prefetched.push((key, lsn, value)); - - while let Some(y) = all_values_iter.next() { - if let Ok((key, lsn, value)) = y { - samples.push(Value::ser(&value)?); - prefetched.push((key, lsn, value)); - if samples.len() == config::ZSTD_MAX_SAMPLES { - break; - } - } else { - break; - } - } - - let dictionary = if samples.len() >= config::ZSTD_MIN_SAMPLES { - zstd::dict::from_samples(&samples, config::ZSTD_MAX_DICTIONARY_SIZE)? - } else { - Vec::new() - }; - - writer = Some(DeltaLayerWriter::new( - self.conf, - self.timeline_id, - self.tenant_id, - key, - lsn_range.clone(), - dictionary, - )?); - - // Number of sample is relatively small, so we should not exceed target file limit - for (key, lsn, value) in prefetched { - writer.as_mut().unwrap().put_value(key, lsn, value)?; - prev_key = Some(key); - } - } else { - writer.as_mut().unwrap().put_value(key, lsn, value)?; - prev_key = Some(key); - } - } - if let Some(writer) = writer { - new_layers.push(writer.finish(prev_key.unwrap().next())?); } - // Sync layers - if !new_layers.is_empty() { - let mut layer_paths: Vec = new_layers.iter().map(|l| l.path()).collect(); - - // also sync the directory - layer_paths.push(self.conf.timeline_path(&self.timeline_id, &self.tenant_id)); - - // Fsync all the layer files and directory using multiple threads to - // minimize latency. - par_fsync::par_fsync(&layer_paths)?; - - layer_paths.pop().unwrap(); + if writer.is_none() { + writer = Some(DeltaLayerWriter::new( + self.conf, + self.timeline_id, + self.tenant_id, + key, + lsn_range.clone(), + )?); } + + writer.as_mut().unwrap().put_value(key, lsn, value)?; + prev_key = Some(key); } + if let Some(writer) = writer { + new_layers.push(writer.finish(prev_key.unwrap().next())?); + } + + // Sync layers + if !new_layers.is_empty() { + let mut layer_paths: Vec = new_layers.iter().map(|l| l.path()).collect(); + + // also sync the directory + layer_paths.push(self.conf.timeline_path(&self.timeline_id, &self.tenant_id)); + + // Fsync all the layer files and directory using multiple threads to + // minimize latency. + par_fsync::par_fsync(&layer_paths)?; + + layer_paths.pop().unwrap(); + } + let mut layers = self.layers.write().unwrap(); let mut new_layer_paths = HashSet::with_capacity(new_layers.len()); for l in new_layers { diff --git a/pageserver/src/layered_repository/delta_layer.rs b/pageserver/src/layered_repository/delta_layer.rs index f94b65b25e..6fa870bcfe 100644 --- a/pageserver/src/layered_repository/delta_layer.rs +++ b/pageserver/src/layered_repository/delta_layer.rs @@ -23,6 +23,24 @@ //! "values" part. The actual page images and WAL records are stored in the //! "values" part. //! +//! # Compression +//! +//! Each value is stored as a Blob, which can optionally be compressed. Compression +//! is done by ZStandard, in dictionary mode, which gives pretty good compression +//! ratio even for small inputs, like WAL records. +//! +//! The dictionary is built separately for each delta layer file, and stored in +//! the file itself. +//! +//! TODO: The ZStandard format includes constant 4-byte "magic bytes" in the beginning +//! of each compressed block. With small values like WAL records, that's pretty wasteful. +//! We could disable those bytes by setting the `include_magibytes' flag to false, +//! but as of this writing that's considered experimental in the zstd crate, and the +//! zstd::bulk::Decompressor::upper_bound() function doesn't work without the magic bytes +//! so we would have to find a different way of allocating the decompression buffer if +//! we did that. +//! + use crate::config; use crate::config::PageServerConf; use crate::layered_repository::blob_io::{BlobCursor, BlobWriter, WriteBlobWriter}; @@ -54,8 +72,6 @@ use utils::{ zid::{ZTenantId, ZTimelineId}, }; -const DICTIONARY_OFFSET: u64 = PAGE_SZ as u64; - /// /// Header stored in the beginning of the file /// @@ -77,6 +93,8 @@ struct Summary { index_start_blk: u32, /// Block within the 'index', where the B-tree root page is stored index_root_blk: u32, + + dictionary_offset: u64, } impl From<&DeltaLayer> for Summary { @@ -92,33 +110,46 @@ impl From<&DeltaLayer> for Summary { index_start_blk: 0, index_root_blk: 0, + + dictionary_offset: 0, } } } -// Flag indicating that this version initialize the page -const WILL_INIT: u64 = 1; - /// -/// Struct representing reference to BLOB in layers. Reference contains BLOB -/// offset, and for WAL records it also contains `will_init` flag. The flag -/// helps to determine the range of records that needs to be applied, without -/// reading/deserializing records themselves. +/// Struct representing reference to BLOB in the file. The reference +/// contains the offset to the BLOB within the file, a flag indicating +/// if it's compressed or not, and also the `will_init` flag. The flag +/// helps to determine the range of records that needs to be applied, +/// without reading/deserializing records themselves. /// #[derive(Debug, Serialize, Deserialize, Copy, Clone)] struct BlobRef(u64); +/// Flag indicating that this blob is compressed +const BLOB_COMPRESSED: u64 = 1; + +/// Flag indicating that this version initializes the page +const WILL_INIT: u64 = 2; + impl BlobRef { + pub fn compressed(&self) -> bool { + (self.0 & BLOB_COMPRESSED) != 0 + } + pub fn will_init(&self) -> bool { (self.0 & WILL_INIT) != 0 } pub fn pos(&self) -> u64 { - self.0 >> 1 + self.0 >> 2 } - pub fn new(pos: u64, will_init: bool) -> BlobRef { - let mut blob_ref = pos << 1; + pub fn new(pos: u64, compressed: bool, will_init: bool) -> BlobRef { + let mut blob_ref = pos << 2; + if compressed { + blob_ref |= BLOB_COMPRESSED; + } if will_init { blob_ref |= WILL_INIT; } @@ -201,6 +232,33 @@ pub struct DeltaLayerInner { prepared_dictionary: Option>, // None if not loaded } +impl DeltaLayerInner { + // Create a new Decompressor, using the prepared dictionary + fn create_decompressor(&self) -> Result>> { + if let Some(dict) = &self.prepared_dictionary { + let decompressor = zstd::bulk::Decompressor::with_prepared_dictionary(dict)?; + Ok(Some(decompressor)) + } else { + Ok(None) + } + } + + // Create a new Decompressor, without using the prepared dictionary. + // + // For the cases that you cannot use 'create_decompressor', if the + // Decompressor needs to outlive 'self'. + fn create_decompressor_not_prepared( + &self, + ) -> Result>> { + if !self.dictionary.is_empty() { + let decompressor = zstd::bulk::Decompressor::with_dictionary(&self.dictionary)?; + Ok(Some(decompressor)) + } else { + Ok(None) + } + } +} + impl Layer for DeltaLayer { fn get_tenant_id(&self) -> ZTenantId { self.tenantid @@ -240,12 +298,7 @@ impl Layer for DeltaLayer { { // Open the file and lock the metadata in memory let inner = self.load()?; - let mut decompressor = match &inner.prepared_dictionary { - Some(dictionary) => Some(zstd::bulk::Decompressor::with_prepared_dictionary( - dictionary, - )?), - None => None, - }; + let mut decompressor = inner.create_decompressor()?; let mut decompress_buf = Vec::new(); // Scan the page versions backwards, starting from `lsn`. @@ -257,7 +310,7 @@ impl Layer for DeltaLayer { ); let search_key = DeltaKey::from_key_lsn(&key, Lsn(lsn_range.end.0 - 1)); - let mut offsets: Vec<(Lsn, u64)> = Vec::new(); + let mut blob_refs: Vec<(Lsn, BlobRef)> = Vec::new(); tree_reader.visit(&search_key.0, VisitDirection::Backwards, |key, value| { let blob_ref = BlobRef(value); @@ -265,31 +318,36 @@ impl Layer for DeltaLayer { return false; } let entry_lsn = DeltaKey::extract_lsn_from_buf(key); - offsets.push((entry_lsn, blob_ref.pos())); + blob_refs.push((entry_lsn, blob_ref)); !blob_ref.will_init() })?; // Ok, 'offsets' now contains the offsets of all the entries we need to read let mut cursor = file.block_cursor(); - for (entry_lsn, pos) in offsets { - let buf = cursor.read_blob(pos).with_context(|| { + for (entry_lsn, blob_ref) in blob_refs { + let buf = cursor.read_blob(blob_ref.pos()).with_context(|| { format!( "Failed to read blob from virtual file {}", file.file.path.display() ) })?; - let val = if let Some(decompressor) = &mut decompressor { - let decompressed_max_len = zstd::bulk::Decompressor::upper_bound(&buf) - .ok_or_else(|| anyhow!("could not get decompressed length"))?; - decompress_buf.clear(); - decompress_buf.reserve(decompressed_max_len); - let _ = decompressor.decompress_to_buffer(&buf, &mut decompress_buf)?; - Value::des(&decompress_buf) + let uncompressed_bytes = if blob_ref.compressed() { + if let Some(ref mut decompressor) = decompressor { + let decompressed_max_len = zstd::bulk::Decompressor::upper_bound(&buf) + .ok_or_else(|| anyhow!("could not get decompressed length"))?; + decompress_buf.clear(); + decompress_buf.reserve(decompressed_max_len); + let _ = decompressor.decompress_to_buffer(&buf, &mut decompress_buf)?; + &decompress_buf + } else { + bail!("blob is compressed, but there was no dictionary"); + } } else { - Value::des(&buf) - } - .with_context(|| { + &buf + }; + + let val = Value::des(uncompressed_bytes).with_context(|| { format!( "Failed to deserialize file blob from virtual file {}", file.file.path.display() @@ -382,12 +440,7 @@ impl Layer for DeltaLayer { tree_reader.dump()?; let mut cursor = file.block_cursor(); - let mut decompressor = match &inner.prepared_dictionary { - Some(dictionary) => Some(zstd::bulk::Decompressor::with_prepared_dictionary( - dictionary, - )?), - None => None, - }; + let mut decompressor = inner.create_decompressor()?; let mut decompress_buf = Vec::new(); // A subroutine to dump a single blob @@ -398,33 +451,38 @@ impl Layer for DeltaLayer { file.file.path.display() ) })?; - let decompressed_buf = if let Some(decompressor) = &mut decompressor { - let decompressed_max_len = zstd::bulk::Decompressor::upper_bound(&buf) - .ok_or_else(|| anyhow!("could not get decompressed length"))?; - decompress_buf.clear(); - decompress_buf.reserve(decompressed_max_len); - let _ = decompressor.decompress_to_buffer(&buf, &mut decompress_buf); - &decompress_buf + + let uncompressed_bytes = if blob_ref.compressed() { + if let Some(ref mut decompressor) = decompressor { + let decompressed_max_len = zstd::bulk::Decompressor::upper_bound(&buf) + .ok_or_else(|| anyhow!("could not get decompressed length"))?; + decompress_buf.clear(); + decompress_buf.reserve(decompressed_max_len); + let _ = decompressor.decompress_to_buffer(&buf, &mut decompress_buf)?; + &decompress_buf + } else { + bail!("blob is compressed, but there was no dictionary"); + } } else { &buf }; - let val = Value::des(decompressed_buf) - .with_context(|| { - format!( - "Failed to deserialize file blob from virtual file {}", - file.file.path.display() - ) - })?; + let val = Value::des(uncompressed_bytes).with_context(|| { + format!( + "Failed to deserialize file blob from virtual file {}", + file.file.path.display() + ) + })?; let desc = match val { Value::Image(img) => { - format!(" img {} bytes", img.len()) + format!("img {} bytes, {} compressed", img.len(), buf.len()) } Value::WalRecord(rec) => { let wal_desc = walrecord::describe_wal_record(&rec)?; format!( - " rec {} bytes will_init: {} {}", + "rec {} bytes, {} compressed, will_init {}: {}", + uncompressed_bytes.len(), buf.len(), rec.will_init(), wal_desc @@ -521,6 +579,7 @@ impl DeltaLayer { let mut expected_summary = Summary::from(self); expected_summary.index_start_blk = actual_summary.index_start_blk; expected_summary.index_root_blk = actual_summary.index_root_blk; + expected_summary.dictionary_offset = actual_summary.dictionary_offset; if actual_summary != expected_summary { bail!("in-file summary does not match expected summary. actual = {:?} expected = {:?}", actual_summary, expected_summary); } @@ -540,12 +599,11 @@ impl DeltaLayer { } let mut cursor = file.block_cursor(); - inner.dictionary = cursor.read_blob(DICTIONARY_OFFSET)?; - inner.prepared_dictionary = if inner.dictionary.is_empty() { - None - } else { - Some(zstd::dict::DecoderDictionary::copy(&inner.dictionary)) - }; + if actual_summary.dictionary_offset != 0 { + inner.dictionary = cursor.read_blob(actual_summary.dictionary_offset)?; + inner.prepared_dictionary = + Some(zstd::dict::DecoderDictionary::copy(&inner.dictionary)); + } inner.index_start_blk = actual_summary.index_start_blk; inner.index_root_blk = actual_summary.index_root_blk; @@ -637,6 +695,16 @@ impl DeltaLayer { /// /// 3. Call `finish`. /// +/// +/// To train the dictionary for compression, the first ZSTD_MAX_SAMPLES values +/// (or up ZSTD_MAX_SAMPLE_BYTES) are buffered in memory, before writing them +/// to disk. When the "sample buffer" fills up, the buffered values are used +/// to train a zstandard dictionary, which is then used to compress all the +/// buffered values, and all subsequent values. So the dictionary is built +/// based on just the first values, but in practice that usually gives pretty +/// good compression for all subsequent data as well. Things like page and +/// tuple headers are similar across all pages of the same relation. +/// pub struct DeltaLayerWriter { conf: &'static PageServerConf, path: PathBuf, @@ -650,6 +718,12 @@ pub struct DeltaLayerWriter { blob_writer: WriteBlobWriter>, compressor: Option>, + dictionary_offset: u64, + + training: bool, + sample_key_lsn_willinit: Vec<(Key, Lsn, bool)>, + sample_sizes: Vec, + sample_data: Vec, } impl DeltaLayerWriter { @@ -662,7 +736,6 @@ impl DeltaLayerWriter { tenantid: ZTenantId, key_start: Key, lsn_range: Range, - dictionary: Vec, ) -> Result { // Create the file initially with a temporary filename. We don't know // the end key yet, so we cannot form the final filename yet. We will @@ -678,20 +751,9 @@ impl DeltaLayerWriter { )); let mut file = VirtualFile::create(&path)?; // make room for the header block - file.seek(SeekFrom::Start(DICTIONARY_OFFSET))?; + file.seek(SeekFrom::Start(PAGE_SZ as u64))?; let buf_writer = BufWriter::new(file); - let mut blob_writer = WriteBlobWriter::new(buf_writer, DICTIONARY_OFFSET); - - let off = blob_writer.write_blob(&dictionary)?; - assert!(off == DICTIONARY_OFFSET); - let compressor = if dictionary.is_empty() { - None - } else { - Some(zstd::bulk::Compressor::with_dictionary( - config::ZSTD_COMPRESSION_LEVEL, - &dictionary, - )?) - }; + let blob_writer = WriteBlobWriter::new(buf_writer, PAGE_SZ as u64); // Initialize the b-tree index builder let block_buf = BlockBuf::new(); @@ -705,7 +767,13 @@ impl DeltaLayerWriter { lsn_range, tree: tree_builder, blob_writer, - compressor, + compressor: None, + dictionary_offset: 0, + + training: true, + sample_key_lsn_willinit: Vec::new(), + sample_sizes: Vec::new(), + sample_data: Vec::new(), }) } @@ -715,24 +783,116 @@ impl DeltaLayerWriter { /// The values must be appended in key, lsn order. /// pub fn put_value(&mut self, key: Key, lsn: Lsn, val: Value) -> Result<()> { + let blob_content = &Value::ser(&val)?; + + // Are we still accumulating values for training the compression dictionary? + if self.training { + self.put_value_train(key, lsn, val.will_init(), blob_content)?; + + if self.sample_sizes.len() >= config::ZSTD_MAX_SAMPLES + || self.sample_data.len() >= config::ZSTD_MAX_SAMPLE_BYTES + { + self.finish_training()?; + } + } else { + self.put_value_flush(key, lsn, val.will_init(), blob_content)?; + } + Ok(()) + } + + /// Accumulate one key-value pair in the samples buffer + fn put_value_train(&mut self, key: Key, lsn: Lsn, will_init: bool, bytes: &[u8]) -> Result<()> { + assert!(self.training); + self.sample_key_lsn_willinit.push((key, lsn, will_init)); + self.sample_sizes.push(bytes.len()); + self.sample_data.extend_from_slice(bytes); + Ok(()) + } + + /// Train the compression dictionary, and flush out all the accumulated + /// key-value pairs to disk. + fn finish_training(&mut self) -> Result<()> { + assert!(self.training); + assert!(self.sample_sizes.len() == self.sample_key_lsn_willinit.len()); + + // Create the dictionary, if we had enough samples for it. + if self.sample_sizes.len() >= config::ZSTD_MIN_SAMPLES { + let dictionary = zstd::dict::from_continuous( + &self.sample_data, + &self.sample_sizes, + config::ZSTD_MAX_DICTIONARY_SIZE, + )?; + + let off = self.blob_writer.write_blob(&dictionary)?; + self.dictionary_offset = off; + + let compressor = zstd::bulk::Compressor::with_dictionary( + config::ZSTD_COMPRESSION_LEVEL, + &dictionary, + )?; + self.compressor = Some(compressor); + }; + self.training = false; + + // release the memory used by the sample buffers + let sample_key_lsn_willinit = std::mem::take(&mut self.sample_key_lsn_willinit); + let sample_sizes = std::mem::take(&mut self.sample_sizes); + let sample_data = std::mem::take(&mut self.sample_data); + + // Compress and write out all the buffered key-value pairs + let mut buf_idx: usize = 0; + for ((key, lsn, will_init), len) in + itertools::izip!(sample_key_lsn_willinit.iter(), sample_sizes.iter()) + { + let end = buf_idx + len; + self.put_value_flush(*key, *lsn, *will_init, &sample_data[buf_idx..end])?; + buf_idx = end; + } + assert!(buf_idx == sample_data.len()); + + Ok(()) + } + + /// Write a key-value pair to the file, compressing it if applicable. + pub fn put_value_flush( + &mut self, + key: Key, + lsn: Lsn, + will_init: bool, + bytes: &[u8], + ) -> Result<()> { + assert!(!self.training); assert!(self.lsn_range.start <= lsn); - let body = &Value::ser(&val)?; - let off = if let Some(ref mut compressor) = self.compressor { - let compressed = compressor.compress(body)?; - self.blob_writer.write_blob(&compressed)? - } else { - self.blob_writer.write_blob(body)? - }; + let mut blob_content = bytes; + let mut compressed = false; - let blob_ref = BlobRef::new(off, val.will_init()); + // Try to compress the blob + let compressed_bytes; + if let Some(ref mut compressor) = self.compressor { + compressed_bytes = compressor.compress(blob_content)?; + // If compressed version is not any smaller than the original, + // store it uncompressed. + if compressed_bytes.len() < blob_content.len() { + blob_content = &compressed_bytes; + compressed = true; + } + } + // Write it to the file + let off = self.blob_writer.write_blob(blob_content)?; + let blob_ref = BlobRef::new(off, compressed, will_init); + + // And store the reference in the B-tree let delta_key = DeltaKey::from_key_lsn(&key, lsn); self.tree.append(&delta_key.0, blob_ref.0)?; Ok(()) } + /// + /// Return an estimate of the file, if it was finished now. + /// pub fn size(&self) -> u64 { self.blob_writer.size() + self.tree.borrow_writer().size() } @@ -740,7 +900,11 @@ impl DeltaLayerWriter { /// /// Finish writing the delta layer. /// - pub fn finish(self, key_end: Key) -> anyhow::Result { + pub fn finish(mut self, key_end: Key) -> anyhow::Result { + if self.training { + self.finish_training()?; + } + let index_start_blk = ((self.blob_writer.size() + PAGE_SZ as u64 - 1) / PAGE_SZ as u64) as u32; @@ -764,6 +928,7 @@ impl DeltaLayerWriter { lsn_range: self.lsn_range.clone(), index_start_blk, index_root_blk, + dictionary_offset: self.dictionary_offset, }; file.seek(SeekFrom::Start(0))?; Summary::ser_into(&summary, &mut file)?; @@ -860,13 +1025,15 @@ impl<'a> DeltaValueIter<'a> { true }, )?; - let decompressor = if !inner.dictionary.is_empty() { - Some(zstd::bulk::Decompressor::with_dictionary( - &inner.dictionary, - )?) - } else { - None - }; + + // We cannot use inner.create_decompressor() here, because it returns + // a Decompressor with lifetime that depends on 'inner', and that + // doesn't live long enough here. Cannot use the prepared dictionary + // for that reason either. Doesn't matter too much in practice because + // this Iterator is used for bulk operations, and loading the dictionary + // isn't that expensive in comparison. + let decompressor = inner.create_decompressor_not_prepared()?; + let iter = DeltaValueIter { all_offsets, next_idx: 0, @@ -886,16 +1053,31 @@ impl<'a> DeltaValueIter<'a> { let lsn = delta_key.lsn(); let buf = self.reader.read_blob(blob_ref.pos())?; - let val = if let Some(decompressor) = &mut self.decompressor { - let decompressed_max_len = zstd::bulk::Decompressor::upper_bound(&buf) - .ok_or_else(|| anyhow!("could not get decompressed length"))?; - self.decompress_buf.clear(); - self.decompress_buf.reserve(decompressed_max_len); - let _ = decompressor.decompress_to_buffer(&buf, &mut self.decompress_buf)?; - Value::des(&self.decompress_buf) + let uncompressed_bytes = if blob_ref.compressed() { + if let Some(decompressor) = &mut self.decompressor { + let decompressed_max_len = zstd::bulk::Decompressor::upper_bound(&buf) + .ok_or_else(|| { + anyhow!( + "could not get decompressed length at offset {}", + blob_ref.pos() + ) + })?; + self.decompress_buf.clear(); + self.decompress_buf.reserve(decompressed_max_len); + let _ = decompressor.decompress_to_buffer(&buf, &mut self.decompress_buf)?; + &self.decompress_buf + } else { + bail!("blob is compressed, but there was no dictionary"); + } } else { - Value::des(&buf) - }?; + &buf + }; + let val = Value::des(uncompressed_bytes).with_context(|| { + format!( + "Failed to deserialize file blob at offset {}", + blob_ref.pos() + ) + })?; self.next_idx += 1; Ok(Some((key, lsn, val))) } else { diff --git a/pageserver/src/layered_repository/image_layer.rs b/pageserver/src/layered_repository/image_layer.rs index 29b3aee625..df44015365 100644 --- a/pageserver/src/layered_repository/image_layer.rs +++ b/pageserver/src/layered_repository/image_layer.rs @@ -19,6 +19,10 @@ //! layer, and offsets to the other parts. The "index" is a B-tree, //! mapping from Key to an offset in the "values" part. The //! actual page images are stored in the "values" part. +//! +//! Each page image is compressed with ZStandard. See Compression section +//! in the delta_layer.rs for more discussion. Difference from a delta +//! layer is that we don't currently use a dictionary. use crate::config; use crate::config::PageServerConf; use crate::layered_repository::blob_io::{BlobCursor, BlobWriter, WriteBlobWriter}; @@ -90,6 +94,35 @@ impl From<&ImageLayer> for Summary { } } +/// +/// Struct representing reference to BLOB in the file. In an image layer, +/// each blob is an image of the page. It can be compressed or not, and +/// that is stored in low bit of the BlobRef. +/// +#[derive(Debug, Serialize, Deserialize, Copy, Clone)] +struct BlobRef(u64); + +/// Flag indicating that this blob is compressed +const BLOB_COMPRESSED: u64 = 1; + +impl BlobRef { + pub fn compressed(&self) -> bool { + (self.0 & BLOB_COMPRESSED) != 0 + } + + pub fn pos(&self) -> u64 { + self.0 >> 1 + } + + pub fn new(pos: u64, compressed: bool) -> BlobRef { + let mut blob_ref = pos << 1; + if compressed { + blob_ref |= BLOB_COMPRESSED; + } + BlobRef(blob_ref) + } +} + /// /// ImageLayer is the in-memory data structure associated with an on-disk image /// file. We keep an ImageLayer in memory for each file, in the LayerMap. If a @@ -121,6 +154,13 @@ pub struct ImageLayerInner { file: Option>, } +impl ImageLayerInner { + fn create_decompressor(&self) -> Result> { + let decompressor = zstd::bulk::Decompressor::new()?; + Ok(decompressor) + } +} + impl Layer for ImageLayer { fn filename(&self) -> PathBuf { PathBuf::from(self.layer_name().to_string()) @@ -160,21 +200,33 @@ impl Layer for ImageLayer { let inner = self.load()?; + let mut decompressor = inner.create_decompressor()?; + let file = inner.file.as_ref().unwrap(); let tree_reader = DiskBtreeReader::new(inner.index_start_blk, inner.index_root_blk, file); let mut keybuf: [u8; KEY_SIZE] = [0u8; KEY_SIZE]; key.write_to_byte_slice(&mut keybuf); - if let Some(offset) = tree_reader.get(&keybuf)? { - let blob = file.block_cursor().read_blob(offset).with_context(|| { - format!( - "failed to read value from data file {} at offset {}", - self.filename().display(), - offset - ) - })?; - let decompressed = zstd::bulk::decompress(&blob, PAGE_SZ)?; - let value = Bytes::from(decompressed); + if let Some(value) = tree_reader.get(&keybuf)? { + let blob_ref = BlobRef(value); + let blob_content = + file.block_cursor() + .read_blob(blob_ref.pos()) + .with_context(|| { + format!( + "failed to read value from data file {} at offset {}", + self.filename().display(), + blob_ref.pos() + ) + })?; + + let uncompressed_bytes = if blob_ref.compressed() { + decompressor.decompress(&blob_content, blob_content.len())? + } else { + blob_content + }; + + let value = Bytes::from(uncompressed_bytes); reconstruct_state.img = Some((self.lsn, value)); Ok(ValueReconstructResult::Complete) @@ -220,7 +272,17 @@ impl Layer for ImageLayer { tree_reader.dump()?; tree_reader.visit(&[0u8; KEY_SIZE], VisitDirection::Forwards, |key, value| { - println!("key: {} offset {}", hex::encode(key), value); + let blob_ref = BlobRef(value); + println!( + "key: {} offset {}{}", + hex::encode(key), + blob_ref.pos(), + if blob_ref.compressed() { + " (compressed)" + } else { + "" + } + ); true })?; @@ -408,6 +470,8 @@ pub struct ImageLayerWriter { blob_writer: WriteBlobWriter, tree: DiskBtreeBuilder, + + compressor: Option>, } impl ImageLayerWriter { @@ -441,6 +505,12 @@ impl ImageLayerWriter { let block_buf = BlockBuf::new(); let tree_builder = DiskBtreeBuilder::new(block_buf); + // TODO: use a dictionary + let compressor = { + let compressor = zstd::bulk::Compressor::new(config::ZSTD_COMPRESSION_LEVEL)?; + Some(compressor) + }; + let writer = ImageLayerWriter { conf, _path: path, @@ -450,6 +520,7 @@ impl ImageLayerWriter { lsn, tree: tree_builder, blob_writer, + compressor, }; Ok(writer) @@ -462,12 +533,36 @@ impl ImageLayerWriter { /// pub fn put_image(&mut self, key: Key, img: &[u8]) -> Result<()> { ensure!(self.key_range.contains(&key)); - let compressed = zstd::bulk::compress(img, config::ZSTD_COMPRESSION_LEVEL)?; - let off = self.blob_writer.write_blob(&compressed)?; + let mut blob_content = img; + let mut compressed = false; + + // Try to compress the blob + let compressed_bytes; + if let Some(ref mut compressor) = self.compressor { + // Try to compress it + compressed_bytes = compressor.compress(blob_content)?; + + // If compressed version is not any smaller than the original, + // store it uncompressed. This not just an optimization, the + // the decompression assumes that too. That simplifies the + // decompression, because you don't need to jump through any + // hoops to determine how large a buffer you need to hold the + // decompression result. + if compressed_bytes.len() < blob_content.len() { + blob_content = &compressed_bytes; + compressed = true; + } + } + + // Write it to the file + let off = self.blob_writer.write_blob(blob_content)?; + let blob_ref = BlobRef::new(off, compressed); + + // And store the reference in the B-tree let mut keybuf: [u8; KEY_SIZE] = [0u8; KEY_SIZE]; key.write_to_byte_slice(&mut keybuf); - self.tree.append(&keybuf, off)?; + self.tree.append(&keybuf, blob_ref.0)?; Ok(()) } diff --git a/pageserver/src/layered_repository/inmemory_layer.rs b/pageserver/src/layered_repository/inmemory_layer.rs index 20cdb5ec5d..bffb946f7e 100644 --- a/pageserver/src/layered_repository/inmemory_layer.rs +++ b/pageserver/src/layered_repository/inmemory_layer.rs @@ -4,7 +4,6 @@ //! held in an ephemeral file, not in memory. The metadata for each page version, i.e. //! its position in the file, is kept in memory, though. //! -use crate::config; use crate::config::PageServerConf; use crate::layered_repository::blob_io::{BlobCursor, BlobWriter}; use crate::layered_repository::block_io::BlockReader; @@ -323,40 +322,21 @@ impl InMemoryLayer { // rare though, so we just accept the potential latency hit for now. let inner = self.inner.read().unwrap(); - let mut samples: Vec> = Vec::with_capacity(config::ZSTD_MAX_SAMPLES); - let mut buf = Vec::new(); - - let mut keys: Vec<(&Key, &VecMap)> = inner.index.iter().collect(); - keys.sort_by_key(|k| k.0); - - let mut cursor = inner.file.block_cursor(); - - // First learn dictionary - 'train: for (_key, vec_map) in keys.iter() { - // Write all page versions - for (_lsn, pos) in vec_map.as_slice() { - cursor.read_blob_into_buf(*pos, &mut buf)?; - samples.push(buf.clone()); - if samples.len() == config::ZSTD_MAX_SAMPLES { - break 'train; - } - } - } - - let dictionary = if samples.len() >= config::ZSTD_MIN_SAMPLES { - zstd::dict::from_samples(&samples, config::ZSTD_MAX_DICTIONARY_SIZE)? - } else { - Vec::new() - }; let mut delta_layer_writer = DeltaLayerWriter::new( self.conf, self.timelineid, self.tenantid, Key::MIN, self.start_lsn..inner.end_lsn.unwrap(), - dictionary, )?; + let mut buf = Vec::new(); + + let mut cursor = inner.file.block_cursor(); + + let mut keys: Vec<(&Key, &VecMap)> = inner.index.iter().collect(); + keys.sort_by_key(|k| k.0); + for (key, vec_map) in keys.iter() { let key = **key; // Write all page versions