Fix check of format version

pageserver/src/pgdatadir_mapping.rs

@@ -167,7 +167,7 @@ impl Timeline {
             return Ok(0);
         }
 
-        let key = rel_size_to_key(tag);
+        let key = self.rel_size_to_key(tag);
         let mut buf = self.get(key, lsn, ctx).await?;
         let nblocks = buf.get_u32_le();
 
@@ -516,7 +516,7 @@ impl Timeline {
                 if cancel.is_cancelled() {
                     return Err(CalculateLogicalSizeError::Cancelled);
                 }
-                let relsize_key = rel_size_to_key(rel);
+                let relsize_key = self.rel_size_to_key(rel);
                 let mut buf = self
                     .get(relsize_key, lsn, ctx)
                     .await
@@ -550,7 +550,9 @@ impl Timeline {
 
         let mut dbs: Vec<(Oid, Oid)> = dbdir.dbdirs.keys().cloned().collect();
         dbs.sort_unstable();
-        for (spcnode, dbnode) in dbs {
+        for (spcnode, dbnode) in &dbs {
+            let spcnode = *spcnode;
+            let dbnode = *dbnode;
             result.add_key(relmap_file_key(spcnode, dbnode));
             result.add_key(rel_dir_to_key(spcnode, dbnode));
 
@@ -561,12 +563,14 @@ impl Timeline {
                 .collect();
             rels.sort_unstable();
             for rel in rels {
-                let relsize_key = rel_size_to_key(rel);
+                let relsize_key = self.rel_size_to_key(rel);
                 let mut buf = self.get(relsize_key, lsn, ctx).await?;
                 let relsize = buf.get_u32_le();
 
                 result.add_range(rel_block_to_key(rel, 0)..rel_block_to_key(rel, relsize));
-                result.add_key(relsize_key);
+                if self.format_version <= 4 {
+                    result.add_key(relsize_key);
+                }
             }
         }
 
@@ -607,6 +611,22 @@ impl Timeline {
         result.add_key(CONTROLFILE_KEY);
         result.add_key(CHECKPOINT_KEY);
 
+        if self.format_version > 4 {
+            // Store relation metadata
+            for (spcnode, dbnode) in dbs {
+                let mut rels: Vec<RelTag> = self
+                    .list_rels(spcnode, dbnode, lsn, ctx)
+                    .await?
+                    .into_iter()
+                    .collect();
+                rels.sort_unstable();
+                for rel in rels {
+                    let relsize_key = self.rel_size_to_key(rel);
+                    result.add_key(relsize_key);
+                }
+            }
+        }
+
         Ok(result.to_keyspace())
     }
 
@@ -648,6 +668,22 @@ impl Timeline {
         let mut rel_size_cache = self.rel_size_cache.write().unwrap();
         rel_size_cache.remove(tag);
     }
+
+    fn rel_size_to_key(&self, rel: RelTag) -> Key {
+        let field1 = if self.format_version <= 4 {
+            0x00u8
+        } else {
+            0x04u8
+        };
+        Key {
+            field1,
+            field2: rel.spcnode,
+            field3: rel.dbnode,
+            field4: rel.relnode,
+            field5: rel.forknum,
+            field6: 0xffffffff,
+        }
+    }
 }
 
 /// DatadirModification represents an operation to ingest an atomic set of
@@ -890,7 +926,7 @@ impl<'a> DatadirModification<'a> {
         );
 
         // Put size
-        let size_key = rel_size_to_key(rel);
+        let size_key = self.tline.rel_size_to_key(rel);
         let buf = nblocks.to_le_bytes();
         self.put(size_key, Value::Image(Bytes::from(buf.to_vec())));
 
@@ -914,7 +950,7 @@ impl<'a> DatadirModification<'a> {
         anyhow::ensure!(rel.relnode != 0, "invalid relnode");
         let last_lsn = self.tline.get_last_record_lsn();
         if self.tline.get_rel_exists(rel, last_lsn, true, ctx).await? {
-            let size_key = rel_size_to_key(rel);
+            let size_key = self.tline.rel_size_to_key(rel);
             // Fetch the old size first
             let old_size = self.get(size_key, ctx).await?.get_u32_le();
 
@@ -945,7 +981,7 @@ impl<'a> DatadirModification<'a> {
         anyhow::ensure!(rel.relnode != 0, "invalid relnode");
 
         // Put size
-        let size_key = rel_size_to_key(rel);
+        let size_key = self.tline.rel_size_to_key(rel);
         let old_size = self.get(size_key, ctx).await?.get_u32_le();
 
         // only extend relation here. never decrease the size
@@ -977,7 +1013,7 @@ impl<'a> DatadirModification<'a> {
         }
 
         // update logical size
-        let size_key = rel_size_to_key(rel);
+        let size_key = self.tline.rel_size_to_key(rel);
         let old_size = self.get(size_key, ctx).await?.get_u32_le();
         self.pending_nblocks -= old_size as i64;
 
@@ -1291,7 +1327,7 @@ static ZERO_PAGE: Bytes = Bytes::from_static(&[0u8; BLCKSZ as usize]);
 // 00 SPCNODE  DBNODE   RELNODE  FORK BLKNUM
 //
 // RelSize:
-// 00 SPCNODE  DBNODE   RELNODE  FORK FFFFFFFF
+// 04 SPCNODE  DBNODE   RELNODE  FORK FFFFFFFF
 //
 // SlruDir:
 // 01 kind     00000000 00000000 00   00000000
@@ -1375,17 +1411,6 @@ fn rel_block_to_key(rel: RelTag, blknum: BlockNumber) -> Key {
     }
 }
 
-fn rel_size_to_key(rel: RelTag) -> Key {
-    Key {
-        field1: 0x00,
-        field2: rel.spcnode,
-        field3: rel.dbnode,
-        field4: rel.relnode,
-        field5: rel.forknum,
-        field6: 0xffffffff,
-    }
-}
-
 fn rel_key_range(rel: RelTag) -> Range<Key> {
     Key {
         field1: 0x00,

pageserver/src/tenant/metadata.rs

@@ -23,7 +23,7 @@ use crate::config::PageServerConf;
 use crate::virtual_file::VirtualFile;
 
 /// Use special format number to enable backward compatibility.
-const METADATA_FORMAT_VERSION: u16 = 4;
+const METADATA_FORMAT_VERSION: u16 = 5;
 
 /// Previous supported format versions.
 const METADATA_OLD_FORMAT_VERSION: u16 = 3;
@@ -145,7 +145,7 @@ impl TimelineMetadata {
             pg_version: 14, // All timelines created before this version had pg_version 14
         };
 
-        hdr.format_version = METADATA_FORMAT_VERSION;
+        hdr.format_version = 4;
 
         Ok(Self { hdr, body })
     }
@@ -168,7 +168,7 @@ impl TimelineMetadata {
             "metadata checksum mismatch"
         );
 
-        if hdr.format_version != METADATA_FORMAT_VERSION {
+        if hdr.format_version == METADATA_OLD_FORMAT_VERSION {
             // If metadata has the old format,
             // upgrade it and return the result
             TimelineMetadata::upgrade_timeline_metadata(metadata_bytes)
@@ -227,6 +227,10 @@ impl TimelineMetadata {
     pub fn pg_version(&self) -> u32 {
         self.body.pg_version
     }
+
+    pub fn format_version(&self) -> u16 {
+        self.hdr.format_version
+    }
 }
 
 /// Save timeline metadata to file
@@ -380,7 +384,7 @@ mod tests {
         assert_eq!(
             deserialized_metadata.body, expected_metadata.body,
             "Metadata of the old version {} should be upgraded to the latest version {}",
-            METADATA_OLD_FORMAT_VERSION, METADATA_FORMAT_VERSION
+            METADATA_OLD_FORMAT_VERSION, 4
         );
     }
 }

pageserver/src/tenant/timeline.rs

@@ -120,6 +120,7 @@ pub struct Timeline {
     pub timeline_id: TimelineId,
 
     pub pg_version: u32,
+    pub format_version: u16,
 
     pub(super) layers: RwLock<LayerMap<dyn PersistentLayer>>,
 
@@ -1353,6 +1354,7 @@ impl Timeline {
                 timeline_id,
                 tenant_id,
                 pg_version,
+                format_version: metadata.format_version(),
                 layers: RwLock::new(LayerMap::default()),
 
                 walredo_mgr,