Prevent commit_lsn <= flush_lsn violation after a42eba3cd7.

Nothing complained about that yet, but we definitely don't hold at least one assert, so let's keep it this way until better version.
s3 WAL offloading staging review.
2026-07-14 01:20:38 +00:00 · 2022-05-27 14:11:56 +04:00 · 2022-05-27 13:17:10 +04:00
12 changed files with 52 additions and 584 deletions
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -292,9 +292,6 @@ name = "cc"
 version = "1.0.72"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "22a9137b95ea06864e018375b72adfb7db6e6f68cfc8df5a04d00288050485ee"
-dependencies = [
- "jobserver",
-]

 [[package]]
 name = "cexpr"
@@ -1359,15 +1356,6 @@ version = "1.0.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1aab8fc367588b89dcee83ab0fd66b72b50b72fa1904d7095045ace2b0c81c35"

-[[package]]
-name = "jobserver"
-version = "0.1.24"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "af25a77299a7f711a01975c35a6a424eb6862092cc2d6c72c4ed6cbc56dfc1fa"
-dependencies = [
- "libc",
-]
-
 [[package]]
 name = "js-sys"
 version = "0.3.56"
@@ -1843,7 +1831,6 @@ dependencies = [
 "url",
 "utils",
 "workspace_hack",
- "zstd",
 ]

 [[package]]
@@ -3951,32 +3938,3 @@ name = "zeroize"
 version = "1.5.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7c88870063c39ee00ec285a2f8d6a966e5b6fb2becc4e8dac77ed0d370ed6006"
-
-[[package]]
-name = "zstd"
-version = "0.11.1+zstd.1.5.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "77a16b8414fde0414e90c612eba70985577451c4c504b99885ebed24762cb81a"
-dependencies = [
- "zstd-safe",
-]
-
-[[package]]
-name = "zstd-safe"
-version = "5.0.1+zstd.1.5.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7c12659121420dd6365c5c3de4901f97145b79651fb1d25814020ed2ed0585ae"
-dependencies = [
- "libc",
- "zstd-sys",
-]
-
-[[package]]
-name = "zstd-sys"
-version = "2.0.1+zstd.1.5.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9fd07cbbc53846d9145dbffdf6dd09a7a0aa52be46741825f5c97bdd4f73f12b"
-dependencies = [
- "cc",
- "libc",
-]
--- a/pageserver/Cargo.toml
+++ b/pageserver/Cargo.toml
@@ -54,9 +54,6 @@ crossbeam-utils = "0.8.5"
 fail = "0.5.0"
 git-version = "0.3.5"

-# 'experimental' is needed for the `zstd::bulk::Decompressor::upper_bound` function.
-zstd = { version = "0.11.1", features = ["experimental"] }
-
 postgres_ffi = { path = "../libs/postgres_ffi" }
 etcd_broker = { path = "../libs/etcd_broker" }
 metrics = { path = "../libs/metrics" }
--- a/pageserver/src/config.rs
+++ b/pageserver/src/config.rs
@@ -22,12 +22,6 @@ use utils::{
 use crate::layered_repository::TIMELINES_SEGMENT_NAME;
 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 - 4; // make dictionary + BLOB length fit in first page
-pub const ZSTD_COMPRESSION_LEVEL: i32 = 0; // default compression level
-
 pub mod defaults {
    use crate::tenant_config::defaults::*;
    use const_format::formatcp;
--- a/pageserver/src/layered_repository.rs
+++ b/pageserver/src/layered_repository.rs
@@ -2518,7 +2518,7 @@ fn rename_to_backup(path: PathBuf) -> anyhow::Result<()> {
    bail!("couldn't find an unused backup number for {:?}", path)
 }

-pub fn load_metadata(
+fn load_metadata(
    conf: &'static PageServerConf,
    timeline_id: ZTimelineId,
    tenant_id: ZTenantId,
--- a/pageserver/src/layered_repository/delta_layer.rs
+++ b/pageserver/src/layered_repository/delta_layer.rs
@@ -23,25 +23,6 @@
 //! "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};
 use crate::layered_repository::block_io::{BlockBuf, BlockCursor, BlockReader, FileBlockReader};
@@ -55,7 +36,7 @@ use crate::repository::{Key, Value, KEY_SIZE};
 use crate::virtual_file::VirtualFile;
 use crate::walrecord;
 use crate::{DELTA_FILE_MAGIC, STORAGE_FORMAT_VERSION};
-use anyhow::{anyhow, bail, ensure, Context, Result};
+use anyhow::{bail, ensure, Context, Result};
 use rand::{distributions::Alphanumeric, Rng};
 use serde::{Deserialize, Serialize};
 use std::fs;
@@ -94,9 +75,6 @@ struct Summary {
    index_start_blk: u32,
    /// Block within the 'index', where the B-tree root page is stored
    index_root_blk: u32,
-
-    /// Byte offset of the compression dictionary, or 0 if no compression
-    dictionary_offset: u64,
 }

 impl From<&DeltaLayer> for Summary {
@@ -112,46 +90,33 @@ 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 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 `will_init` flag
+/// 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.
 ///
 #[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 >> 2
+        self.0 >> 1
    }

-    pub fn new(pos: u64, compressed: bool, will_init: bool) -> BlobRef {
-        let mut blob_ref = pos << 2;
-        if compressed {
-            blob_ref |= BLOB_COMPRESSED;
-        }
+    pub fn new(pos: u64, will_init: bool) -> BlobRef {
+        let mut blob_ref = pos << 1;
        if will_init {
            blob_ref |= WILL_INIT;
        }
@@ -228,37 +193,6 @@ pub struct DeltaLayerInner {

    /// Reader object for reading blocks from the file. (None if not loaded yet)
    file: Option<FileBlockReader<VirtualFile>>,
-
-    /// Compression dictionary, as raw bytes, and in prepared format ready for use
-    /// for decompression. None if there is no dictionary, or if 'loaded' is false.
-    dictionary: Option<(Vec<u8>, zstd::dict::DecoderDictionary<'static>)>,
-}
-
-impl DeltaLayerInner {
-    // Create a new Decompressor, using the prepared dictionary
-    fn create_decompressor(&self) -> Result<Option<zstd::bulk::Decompressor<'_>>> {
-        if let Some((_, dict)) = &self.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<Option<zstd::bulk::Decompressor<'static>>> {
-        if let Some((dict, _)) = &self.dictionary {
-            let decompressor = zstd::bulk::Decompressor::with_dictionary(dict)?;
-            Ok(Some(decompressor))
-        } else {
-            Ok(None)
-        }
-    }
 }

 impl Layer for DeltaLayer {
@@ -300,8 +234,6 @@ impl Layer for DeltaLayer {
        {
            // Open the file and lock the metadata in memory
            let inner = self.load()?;
-            let mut decompressor = inner.create_decompressor()?;
-            let mut decompress_buf = Vec::new();

            // Scan the page versions backwards, starting from `lsn`.
            let file = inner.file.as_ref().unwrap();
@@ -312,7 +244,7 @@ impl Layer for DeltaLayer {
            );
            let search_key = DeltaKey::from_key_lsn(&key, Lsn(lsn_range.end.0 - 1));

-            let mut blob_refs: Vec<(Lsn, BlobRef)> = Vec::new();
+            let mut offsets: Vec<(Lsn, u64)> = Vec::new();

            tree_reader.visit(&search_key.0, VisitDirection::Backwards, |key, value| {
                let blob_ref = BlobRef(value);
@@ -323,36 +255,21 @@ impl Layer for DeltaLayer {
                if entry_lsn < lsn_range.start {
                    return false;
                }
-                blob_refs.push((entry_lsn, blob_ref));
+                offsets.push((entry_lsn, blob_ref.pos()));

                !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, blob_ref) in blob_refs {
-                let buf = cursor.read_blob(blob_ref.pos()).with_context(|| {
+            for (entry_lsn, pos) in offsets {
+                let buf = cursor.read_blob(pos).with_context(|| {
                    format!(
                        "Failed to read blob from virtual file {}",
                        file.file.path.display()
                    )
                })?;
-                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(uncompressed_bytes).with_context(|| {
+                let val = Value::des(&buf).with_context(|| {
                    format!(
                        "Failed to deserialize file blob from virtual file {}",
                        file.file.path.display()
@@ -430,6 +347,7 @@ impl Layer for DeltaLayer {
        }

        let inner = self.load()?;
+
        println!(
            "index_start_blk: {}, root {}",
            inner.index_start_blk, inner.index_root_blk
@@ -445,49 +363,19 @@ impl Layer for DeltaLayer {
        tree_reader.dump()?;

        let mut cursor = file.block_cursor();
-        let mut decompressor = inner.create_decompressor()?;
-        let mut decompress_buf = Vec::new();

        // A subroutine to dump a single blob
        let mut dump_blob = |blob_ref: BlobRef| -> anyhow::Result<String> {
-            let buf = cursor.read_blob(blob_ref.pos()).with_context(|| {
-                format!(
-                    "Failed to read blob from virtual file {}",
-                    file.file.path.display()
-                )
-            })?;
-
-            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(uncompressed_bytes).with_context(|| {
-                format!(
-                    "Failed to deserialize file blob from virtual file {}",
-                    file.file.path.display()
-                )
-            })?;
-
+            let buf = cursor.read_blob(blob_ref.pos())?;
+            let val = Value::des(&buf)?;
            let desc = match val {
                Value::Image(img) => {
-                    format!("img {} bytes, {} compressed", img.len(), buf.len())
+                    format!(" img {} bytes", img.len())
                }
                Value::WalRecord(rec) => {
                    let wal_desc = walrecord::describe_wal_record(&rec)?;
                    format!(
-                        "rec {} bytes, {} compressed, will_init {}: {}",
-                        uncompressed_bytes.len(),
+                        " rec {} bytes will_init: {} {}",
                        buf.len(),
                        rec.will_init(),
                        wal_desc
@@ -606,7 +494,6 @@ 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);
                }
@@ -625,13 +512,6 @@ impl DeltaLayer {
            }
        }

-        // Load and prepare the dictionary, if any
-        if actual_summary.dictionary_offset != 0 {
-            let mut cursor = file.block_cursor();
-            let dict = cursor.read_blob(actual_summary.dictionary_offset)?;
-            let prepared_dict = zstd::dict::DecoderDictionary::copy(&dict);
-            inner.dictionary = Some((dict, prepared_dict));
-        }
        inner.index_start_blk = actual_summary.index_start_blk;
        inner.index_root_blk = actual_summary.index_root_blk;

@@ -657,7 +537,6 @@ impl DeltaLayer {
            inner: RwLock::new(DeltaLayerInner {
                loaded: false,
                file: None,
-                dictionary: None,
                index_start_blk: 0,
                index_root_blk: 0,
            }),
@@ -685,7 +564,6 @@ impl DeltaLayer {
            inner: RwLock::new(DeltaLayerInner {
                loaded: false,
                file: None,
-                dictionary: None,
                index_start_blk: 0,
                index_root_blk: 0,
            }),
@@ -721,16 +599,6 @@ 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,
@@ -743,13 +611,6 @@ pub struct DeltaLayerWriter {
    tree: DiskBtreeBuilder<BlockBuf, DELTA_KEY_SIZE>,

    blob_writer: WriteBlobWriter<BufWriter<VirtualFile>>,
-    compressor: Option<zstd::bulk::Compressor<'static>>,
-    dictionary_offset: u64,
-
-    training: bool,
-    sample_key_lsn_willinit: Vec<(Key, Lsn, bool)>,
-    sample_sizes: Vec<usize>,
-    sample_data: Vec<u8>,
 }

 impl DeltaLayerWriter {
@@ -780,6 +641,7 @@ impl DeltaLayerWriter {
        // Initialize the b-tree index builder
        let block_buf = BlockBuf::new();
        let tree_builder = DiskBtreeBuilder::new(block_buf);
+
        Ok(DeltaLayerWriter {
            conf,
            path,
@@ -789,13 +651,6 @@ impl DeltaLayerWriter {
            lsn_range,
            tree: tree_builder,
            blob_writer,
-            compressor: None,
-            dictionary_offset: 0,
-
-            training: true,
-            sample_key_lsn_willinit: Vec::new(),
-            sample_sizes: Vec::new(),
-            sample_data: Vec::new(),
        })
    }

@@ -805,122 +660,18 @@ 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 there weren't enough samples, we don't do any compression at
-        // all. Possibly we could still benefit from compression; for example
-        // if you have only one gigantic value in a single layer, it would
-        // still be good to compress that, without a dictionary. But we don't
-        // do that currently.
-        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 mut blob_content = bytes;
-        let mut compressed = false;
+        let off = self.blob_writer.write_blob(&Value::ser(&val)?)?;

-        // 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;
-            }
-        }
+        let blob_ref = BlobRef::new(off, val.will_init());

-        // 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()
    }
@@ -928,11 +679,7 @@ impl DeltaLayerWriter {
    ///
    /// Finish writing the delta layer.
    ///
-    pub fn finish(mut self, key_end: Key) -> anyhow::Result<DeltaLayer> {
-        if self.training {
-            self.finish_training()?;
-        }
-
+    pub fn finish(self, key_end: Key) -> anyhow::Result<DeltaLayer> {
        let index_start_blk =
            ((self.blob_writer.size() + PAGE_SZ as u64 - 1) / PAGE_SZ as u64) as u32;

@@ -956,7 +703,6 @@ 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)?;
@@ -973,7 +719,6 @@ impl DeltaLayerWriter {
            inner: RwLock::new(DeltaLayerInner {
                loaded: false,
                file: None,
-                dictionary: None,
                index_start_blk,
                index_root_blk,
            }),
@@ -1013,9 +758,6 @@ struct DeltaValueIter<'a> {
    all_offsets: Vec<(DeltaKey, BlobRef)>,
    next_idx: usize,
    reader: BlockCursor<Adapter<'a>>,
-    decompressor: Option<zstd::bulk::Decompressor<'a>>,
-
-    decompress_buf: Vec<u8>,
 }

 struct Adapter<'a>(RwLockReadGuard<'a, DeltaLayerInner>);
@@ -1055,20 +797,10 @@ impl<'a> DeltaValueIter<'a> {
            },
        )?;

-        // 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,
            reader: BlockCursor::new(Adapter(inner)),
-            decompressor,
-            decompress_buf: Vec::new(),
        };

        Ok(iter)
@@ -1082,31 +814,7 @@ impl<'a> DeltaValueIter<'a> {
            let lsn = delta_key.lsn();

            let buf = self.reader.read_blob(blob_ref.pos())?;
-            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 {
-                &buf
-            };
-            let val = Value::des(uncompressed_bytes).with_context(|| {
-                format!(
-                    "Failed to deserialize file blob at offset {}",
-                    blob_ref.pos()
-                )
-            })?;
+            let val = Value::des(&buf)?;
            self.next_idx += 1;
            Ok(Some((key, lsn, val)))
        } else {
--- a/pageserver/src/layered_repository/image_layer.rs
+++ b/pageserver/src/layered_repository/image_layer.rs
@@ -19,11 +19,6 @@
 //! 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 for image layers.
-use crate::config;
 use crate::config::PageServerConf;
 use crate::layered_repository::blob_io::{BlobCursor, BlobWriter, WriteBlobWriter};
 use crate::layered_repository::block_io::{BlockBuf, BlockReader, FileBlockReader};
@@ -95,35 +90,6 @@ 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
@@ -155,13 +121,6 @@ pub struct ImageLayerInner {
    file: Option<FileBlockReader<VirtualFile>>,
 }

-impl ImageLayerInner {
-    fn create_decompressor(&self) -> Result<zstd::bulk::Decompressor<'_>> {
-        let decompressor = zstd::bulk::Decompressor::new()?;
-        Ok(decompressor)
-    }
-}
-
 impl Layer for ImageLayer {
    fn filename(&self) -> PathBuf {
        PathBuf::from(self.layer_name().to_string())
@@ -201,33 +160,20 @@ 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(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, PAGE_SZ)?
-            } else {
-                blob_content
-            };
-
-            let value = Bytes::from(uncompressed_bytes);
+        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 value = Bytes::from(blob);

            reconstruct_state.img = Some((self.lsn, value));
            Ok(ValueReconstructResult::Complete)
@@ -273,17 +219,7 @@ impl Layer for ImageLayer {
        tree_reader.dump()?;

        tree_reader.visit(&[0u8; KEY_SIZE], VisitDirection::Forwards, |key, value| {
-            let blob_ref = BlobRef(value);
-            println!(
-                "key: {} offset {}{}",
-                hex::encode(key),
-                blob_ref.pos(),
-                if blob_ref.compressed() {
-                    " (compressed)"
-                } else {
-                    ""
-                }
-            );
+            println!("key: {} offset {}", hex::encode(key), value);
            true
        })?;

@@ -487,8 +423,6 @@ pub struct ImageLayerWriter {

    blob_writer: WriteBlobWriter<VirtualFile>,
    tree: DiskBtreeBuilder<BlockBuf, KEY_SIZE>,
-
-    compressor: Option<zstd::bulk::Compressor<'static>>,
 }

 impl ImageLayerWriter {
@@ -520,12 +454,6 @@ 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,
@@ -535,7 +463,6 @@ impl ImageLayerWriter {
            lsn,
            tree: tree_builder,
            blob_writer,
-            compressor,
        };

        Ok(writer)
@@ -548,37 +475,11 @@ impl ImageLayerWriter {
    ///
    pub fn put_image(&mut self, key: Key, img: &[u8]) -> Result<()> {
        ensure!(self.key_range.contains(&key));
+        let off = self.blob_writer.write_blob(img)?;

-        let mut blob_content = img;
-        let mut compressed = false;
-
-        // Try to compress the blob
-        let compressed_bytes;
-        if blob_content.len() <= PAGE_SZ {
-            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. 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, blob_ref.0)?;
+        self.tree.append(&keybuf, off)?;

        Ok(())
    }
--- a/pageserver/src/lib.rs
+++ b/pageserver/src/lib.rs
@@ -37,7 +37,7 @@ use pgdatadir_mapping::DatadirTimeline;
 /// This is embedded in the metadata file, and also in the header of all the
 /// layer files. If you make any backwards-incompatible changes to the storage
 /// format, bump this!
-pub const STORAGE_FORMAT_VERSION: u16 = 4;
+pub const STORAGE_FORMAT_VERSION: u16 = 3;

 // Magic constants used to identify different kinds of files
 pub const IMAGE_FILE_MAGIC: u16 = 0x5A60;
--- a/pageserver/src/tenant_mgr.rs
+++ b/pageserver/src/tenant_mgr.rs
@@ -2,7 +2,7 @@
 //! page server.

 use crate::config::PageServerConf;
-use crate::layered_repository::{load_metadata, LayeredRepository};
+use crate::layered_repository::LayeredRepository;
 use crate::pgdatadir_mapping::DatadirTimeline;
 use crate::repository::{Repository, TimelineSyncStatusUpdate};
 use crate::storage_sync::index::RemoteIndex;
@@ -22,7 +22,6 @@ use std::collections::HashMap;
 use std::fmt;
 use std::sync::Arc;
 use tracing::*;
-use utils::lsn::Lsn;

 use utils::zid::{ZTenantId, ZTimelineId};

@@ -400,26 +399,6 @@ pub fn list_tenants() -> Vec<TenantInfo> {
        .collect()
 }

-/// Check if a given timeline is "broken" \[1\].
-/// The function returns an error if the timeline is "broken".
-///
-/// \[1\]: it's not clear now how should we classify a timeline as broken.
-/// A timeline is categorized as broken when any of following conditions is true:
-/// - failed to load the timeline's metadata
-/// - the timeline's disk consistent LSN is zero
-fn check_broken_timeline(repo: &LayeredRepository, timeline_id: ZTimelineId) -> anyhow::Result<()> {
-    let metadata = load_metadata(repo.conf, timeline_id, repo.tenant_id())
-        .context("failed to load metadata")?;
-
-    // A timeline with zero disk consistent LSN can happen when the page server
-    // failed to checkpoint the timeline import data when creating that timeline.
-    if metadata.disk_consistent_lsn() == Lsn::INVALID {
-        bail!("Timeline {timeline_id} has a zero disk consistent LSN.");
-    }
-
-    Ok(())
-}
-
 fn init_local_repository(
    conf: &'static PageServerConf,
    tenant_id: ZTenantId,
@@ -435,13 +414,7 @@ fn init_local_repository(
        match init_status {
            LocalTimelineInitStatus::LocallyComplete => {
                debug!("timeline {timeline_id} for tenant {tenant_id} is locally complete, registering it in repository");
-                if let Err(err) = check_broken_timeline(&repo, timeline_id) {
-                    info!(
-                        "Found a broken timeline {timeline_id} (err={err:?}), skip registering it in repository"
-                    );
-                } else {
-                    status_updates.insert(timeline_id, TimelineSyncStatusUpdate::Downloaded);
-                }
+                status_updates.insert(timeline_id, TimelineSyncStatusUpdate::Downloaded);
            }
            LocalTimelineInitStatus::NeedsSync => {
                debug!(
--- a/pageserver/src/timelines.rs
+++ b/pageserver/src/timelines.rs
@@ -285,9 +285,7 @@ fn bootstrap_timeline<R: Repository>(
 ) -> Result<()> {
    let _enter = info_span!("bootstrapping", timeline = %tli, tenant = %tenantid).entered();

-    let initdb_path = conf
-        .tenant_path(&tenantid)
-        .join(format!("tmp-timeline-{}", tli));
+    let initdb_path = conf.tenant_path(&tenantid).join("tmp");

    // Init temporarily repo to get bootstrap data
    run_initdb(conf, &initdb_path)?;
@@ -302,11 +300,6 @@ fn bootstrap_timeline<R: Repository>(
    let timeline = repo.create_empty_timeline(tli, lsn)?;
    let mut page_tline: DatadirTimeline<R> = DatadirTimeline::new(timeline, u64::MAX);
    import_datadir::import_timeline_from_postgres_datadir(&pgdata_path, &mut page_tline, lsn)?;
-
-    fail::fail_point!("before-checkpoint-new-timeline", |_| {
-        bail!("failpoint before-checkpoint-new-timeline");
-    });
-
    page_tline.tline.checkpoint(CheckpointConfig::Forced)?;

    info!(
--- a/safekeeper/src/safekeeper.rs
+++ b/safekeeper/src/safekeeper.rs
@@ -576,13 +576,16 @@ where
        self.state
            .acceptor_state
            .term_history
-            .up_to(self.wal_store.flush_lsn())
+            .up_to(self.flush_lsn())
    }

    pub fn get_epoch(&self) -> Term {
-        self.state
-            .acceptor_state
-            .get_epoch(self.wal_store.flush_lsn())
+        self.state.acceptor_state.get_epoch(self.flush_lsn())
+    }
+
+    /// wal_store wrapper avoiding commit_lsn <= flush_lsn violation when we don't have WAL yet.
+    fn flush_lsn(&self) -> Lsn {
+        max(self.wal_store.flush_lsn(), self.state.timeline_start_lsn)
    }

    /// Process message from proposer and possibly form reply. Concurrent
@@ -671,7 +674,7 @@ where
        let mut resp = VoteResponse {
            term: self.state.acceptor_state.term,
            vote_given: false as u64,
-            flush_lsn: self.wal_store.flush_lsn(),
+            flush_lsn: self.flush_lsn(),
            truncate_lsn: self.state.peer_horizon_lsn,
            term_history: self.get_term_history(),
            timeline_start_lsn: self.state.timeline_start_lsn,
@@ -703,7 +706,7 @@ where
    fn append_response(&self) -> AppendResponse {
        let ar = AppendResponse {
            term: self.state.acceptor_state.term,
-            flush_lsn: self.wal_store.flush_lsn(),
+            flush_lsn: self.flush_lsn(),
            commit_lsn: self.state.commit_lsn,
            // will be filled by the upper code to avoid bothering safekeeper
            hs_feedback: HotStandbyFeedback::empty(),
@@ -770,7 +773,7 @@ where

    /// Advance commit_lsn taking into account what we have locally
    pub fn update_commit_lsn(&mut self) -> Result<()> {
-        let commit_lsn = min(self.global_commit_lsn, self.wal_store.flush_lsn());
+        let commit_lsn = min(self.global_commit_lsn, self.flush_lsn());
        assert!(commit_lsn >= self.inmem.commit_lsn);

        self.inmem.commit_lsn = commit_lsn;
--- a/test_runner/batch_others/test_broken_timeline.py
+++ b/test_runner/batch_others/test_broken_timeline.py
@@ -1,7 +1,6 @@
 import pytest
-import concurrent.futures
 from contextlib import closing
-from fixtures.zenith_fixtures import ZenithEnvBuilder, ZenithEnv
+from fixtures.zenith_fixtures import ZenithEnvBuilder
 from fixtures.log_helper import log
 import os

@@ -79,37 +78,3 @@ def test_broken_timeline(zenith_env_builder: ZenithEnvBuilder):
        with pytest.raises(Exception, match="Cannot load local timeline") as err:
            pg.start()
        log.info(f'compute startup failed as expected: {err}')
-
-
-def test_create_multiple_timelines_parallel(zenith_simple_env: ZenithEnv):
-    env = zenith_simple_env
-
-    tenant_id, _ = env.zenith_cli.create_tenant()
-
-    with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
-        futures = [
-            executor.submit(env.zenith_cli.create_timeline,
-                            f"test-create-multiple-timelines-{i}",
-                            tenant_id) for i in range(4)
-        ]
-        for future in futures:
-            future.result()
-
-
-def test_fix_broken_timelines_on_startup(zenith_simple_env: ZenithEnv):
-    env = zenith_simple_env
-
-    tenant_id, _ = env.zenith_cli.create_tenant()
-
-    # Introduce failpoint when creating a new timeline
-    env.pageserver.safe_psql(f"failpoints before-checkpoint-new-timeline=return")
-    with pytest.raises(Exception, match="before-checkpoint-new-timeline"):
-        _ = env.zenith_cli.create_timeline("test_fix_broken_timelines", tenant_id)
-
-    # Restart the page server
-    env.zenith_cli.pageserver_stop(immediate=True)
-    env.zenith_cli.pageserver_start()
-
-    # Check that the "broken" timeline is not loaded
-    timelines = env.zenith_cli.list_timelines(tenant_id)
-    assert len(timelines) == 1
--- a/test_runner/performance/test_compression.py
+++ b/test_runner/performance/test_compression.py
@@ -1,24 +0,0 @@
-# Test sequential scan speed
-#
-from contextlib import closing
-from dataclasses import dataclass
-from fixtures.zenith_fixtures import ZenithEnv
-from fixtures.log_helper import log
-from fixtures.benchmark_fixture import MetricReport, ZenithBenchmarker
-from fixtures.compare_fixtures import PgCompare
-import pytest
-
-
-@pytest.mark.parametrize('rows', [pytest.param(10000000)])
-def test_compression(zenith_with_baseline: PgCompare, rows: int):
-    env = zenith_with_baseline
-
-    with closing(env.pg.connect()) as conn:
-        with conn.cursor() as cur:
-            with env.record_duration('insert'):
-                cur.execute(
-                    f'create table t as select generate_series(1,{rows}) as pk,(random()*10)::bigint as r10,(random()*100)::bigint as r100,(random()*1000)::bigint as r1000,(random()*10000)::bigint as r10000'
-                )
-            cur.execute("vacuum t")
-            with env.record_duration('select'):
-                cur.execute('select sum(r100) from t')