Represent absent pages in Table and snapshots

crates/engine/src/metrics.rs

@@ -28,6 +28,12 @@ metrics_group!(
         #[labels(db: Identity, kind: str)]
         pub replay_snapshot_num_objects_read: IntGaugeVec,
 
+        #[name = spacetime_replay_snapshot_num_absent_pages]
+        #[help = "Number of pages marked absent and skipped during snapshot replay"]
+        // Not labeled with `kind` as blobs and metadata will never have absent files.
+        #[labels(db: Identity)]
+        pub replay_snapshot_num_absent_pages: IntGaugeVec,
+
         #[name = spacetime_replay_snapshot_bytes_read_from_disk]
         #[help = "Number of snapshot bytes read from disk during replay"]
         #[labels(db: Identity, kind: str)]

crates/engine/src/relational_db.rs

@@ -534,6 +534,11 @@ impl RelationalDB {
 
             let elapsed_time = start.elapsed();
 
+            ENGINE_METRICS
+                .replay_snapshot_num_absent_pages
+                .with_label_values(database_identity)
+                .set(u64_to_i64(snapshot.read_metrics.absent_pages));
+
             for (kind, metrics) in snapshot.read_metrics.iter() {
                 ENGINE_METRICS
                     .replay_snapshot_read_time_seconds

crates/sats/src/proptest.rs

@@ -224,6 +224,12 @@ pub fn generate_typed_row() -> impl Strategy<Value = (ProductType, ProductValue)
     gen_with(generate_row_type(0..=SIZE), generate_product_value)
 }
 
+pub fn generate_two_typed_rows() -> impl Strategy<Value = (ProductType, (ProductValue, ProductValue))> {
+    gen_with(generate_row_type(0..=SIZE), |ty| {
+        (generate_product_value(ty.clone()), generate_product_value(ty))
+    })
+}
+
 pub fn generate_typed_row_vec(
     size: impl Into<SizeRange>,
     num_rows_min: usize,

crates/snapshot/src/lib.rs

@@ -97,6 +97,23 @@ pub struct SnapshotReadMetrics {
     pub metadata: SnapshotReadKindMetrics,
     pub page: SnapshotReadKindMetrics,
     pub blob: SnapshotReadKindMetrics,
+    /// Number of files skipped due to having [`ZERO_HASH_DENOTING_ABSENT_PAGE`] as their hash.
+    ///
+    /// This is only meaningful for pages, so is stored separate from the other metrics.
+    pub absent_pages: u64,
+}
+
+pub type TablePages = Vec<Option<Box<Page>>>;
+
+fn table_present_pages(pages: &[blake3::Hash]) -> impl Iterator<Item = blake3::Hash> {
+    pages
+        .iter()
+        .copied()
+        .filter(|hash| *hash != ZERO_HASH_DENOTING_ABSENT_PAGE)
+}
+
+fn table_num_present_pages(pages: &[blake3::Hash]) -> u64 {
+    table_present_pages(pages).count() as u64
 }
 
 impl SnapshotReadMetrics {
@@ -187,6 +204,8 @@ pub const INVALID_SNAPSHOT_DIR_EXT: &str = "invalid_snapshot";
 /// File extension of snapshots which have been archived
 pub const ARCHIVED_SNAPSHOT_EXT: &str = "archived_snapshot";
 
+pub const ZERO_HASH_DENOTING_ABSENT_PAGE: blake3::Hash = blake3::Hash::from_bytes([0; _]);
+
 #[derive(Clone, Serialize, Deserialize)]
 /// The hash and refcount of a single blob in the blob store.
 struct BlobEntry {
@@ -428,7 +447,13 @@ impl Snapshot {
     ) -> Result<(), SnapshotError> {
         let pages = table
             .iter_pages_with_hashes()
-            .map(|(hash, page)| Self::write_page(object_repo, page, hash, prev_snapshot, counter))
+            .map(|option| {
+                if let Some((hash, page)) = option {
+                    Self::write_page(object_repo, page, hash, prev_snapshot, counter)
+                } else {
+                    Ok(ZERO_HASH_DENOTING_ABSENT_PAGE)
+                }
+            })
             .collect::<Result<Vec<blake3::Hash>, SnapshotError>>()?;
 
         self.tables.push(TableEntry {
@@ -592,40 +617,44 @@ impl Snapshot {
         pages: &[blake3::Hash],
         page_pool: &PagePool,
         metrics: &mut SnapshotReadKindMetrics,
-    ) -> Result<Vec<Box<Page>>, SnapshotError> {
+    ) -> Result<TablePages, SnapshotError> {
         pages
             .iter()
             .map(|hash| {
-                // Read the BSATN bytes of the on-disk page object.
-                let (buf, disk_bytes) =
-                    Self::read_object_with_disk_bytes(object_repo, hash.as_bytes(), ObjectType::Page(*hash))?;
-                metrics.disk_bytes += disk_bytes;
-
-                // Deserialize the bytes into a `Page`.
-                let page = page_pool.take_deserialize_from(&buf);
-                let page = page.map_err(|cause| SnapshotError::Deserialize {
-                    ty: ObjectType::Page(*hash),
-                    source_repo: object_repo.root().to_path_buf(),
-                    cause,
-                })?;
-
-                // Compute the hash of the page.
-                let hash_start = Instant::now();
-                let computed_hash = page.content_hash();
-                metrics.hash_time += hash_start.elapsed();
-
-                // Compare the computed hash to the one recorded in the `Snapshot`,
-                // and fail if they do not match.
-                if *hash != computed_hash {
-                    return Err(SnapshotError::HashMismatch {
+                if *hash == ZERO_HASH_DENOTING_ABSENT_PAGE {
+                    Ok(None)
+                } else {
+                    // Read the BSATN bytes of the on-disk page object.
+                    let (buf, disk_bytes) =
+                        Self::read_object_with_disk_bytes(object_repo, hash.as_bytes(), ObjectType::Page(*hash))?;
+                    metrics.disk_bytes += disk_bytes;
+
+                    // Deserialize the bytes into a `Page`.
+                    let page = page_pool.take_deserialize_from(&buf);
+                    let page = page.map_err(|cause| SnapshotError::Deserialize {
                         ty: ObjectType::Page(*hash),
-                        expected: *hash.as_bytes(),
-                        computed: *computed_hash.as_bytes(),
                         source_repo: object_repo.root().to_path_buf(),
-                    });
-                }
+                        cause,
+                    })?;
+
+                    // Compute the hash of the page.
+                    let hash_start = Instant::now();
+                    let computed_hash = page.content_hash();
+                    metrics.hash_time += hash_start.elapsed();
+
+                    // Compare the computed hash to the one recorded in the `Snapshot`,
+                    // and fail if they do not match.
+                    if *hash != computed_hash {
+                        return Err(SnapshotError::HashMismatch {
+                            ty: ObjectType::Page(*hash),
+                            expected: *hash.as_bytes(),
+                            computed: *computed_hash.as_bytes(),
+                            source_repo: object_repo.root().to_path_buf(),
+                        });
+                    }
 
-                Ok::<Box<Page>, SnapshotError>(page)
+                    Ok::<Option<Box<Page>>, SnapshotError>(Some(page))
+                }
             })
             .collect()
     }
@@ -635,7 +664,7 @@ impl Snapshot {
         TableEntry { table_id, pages }: &TableEntry,
         page_pool: &PagePool,
         metrics: &mut SnapshotReadKindMetrics,
-    ) -> Result<(TableId, Vec<Box<Page>>), SnapshotError> {
+    ) -> Result<(TableId, TablePages), SnapshotError> {
         Ok((
             *table_id,
             Self::reconstruct_one_table_pages(object_repo, pages, page_pool, metrics)?,
@@ -657,7 +686,7 @@ impl Snapshot {
         object_repo: &DirTrie,
         page_pool: &PagePool,
         metrics: &mut SnapshotReadKindMetrics,
-    ) -> Result<BTreeMap<TableId, Vec<Box<Page>>>, SnapshotError> {
+    ) -> Result<BTreeMap<TableId, TablePages>, SnapshotError> {
         self.tables
             .iter()
             .map(|tbl| Self::reconstruct_one_table(object_repo, tbl, page_pool, metrics))
@@ -666,7 +695,12 @@ impl Snapshot {
 
     /// The number of objects in this snapshot, both blobs and pages.
     pub fn total_objects(&self) -> usize {
-        self.blobs.len() + self.tables.iter().map(|table| table.pages.len()).sum::<usize>()
+        self.blobs.len()
+            + self
+                .tables
+                .iter()
+                .map(|table| table_num_present_pages(&table.pages) as usize)
+                .sum::<usize>()
     }
 
     /// Obtain an iterator over the [`blake3::Hash`]es of all objects
@@ -675,7 +709,7 @@ impl Snapshot {
         self.blobs
             .iter()
             .map(|b| blake3::Hash::from_bytes(b.hash.data))
-            .chain(self.tables.iter().flat_map(|t| t.pages.iter().copied()))
+            .chain(self.tables.iter().flat_map(|t| table_present_pages(&t.pages)))
     }
 
     /// Obtain an iterator over the [`PathBuf`]s of all objects
@@ -1041,7 +1075,22 @@ impl SnapshotRepository {
             ..Default::default()
         };
         read_metrics.blob.files = snapshot.blobs.len() as u64;
-        read_metrics.page.files = snapshot.tables.iter().map(|table| table.pages.len() as u64).sum();
+        read_metrics.page.files = snapshot
+            .tables
+            .iter()
+            .map(|table| table_num_present_pages(&table.pages))
+            .sum();
+        read_metrics.absent_pages = snapshot
+            .tables
+            .iter()
+            .map(|table| {
+                table
+                    .pages
+                    .iter()
+                    .filter(|page| **page == ZERO_HASH_DENOTING_ABSENT_PAGE)
+                    .count() as u64
+            })
+            .sum();
 
         if snapshot.magic != MAGIC {
             return Err(SnapshotError::BadMagic {
@@ -1550,7 +1599,7 @@ pub struct ReconstructedSnapshot {
     /// This includes the system tables,
     /// so the schema of user-defined tables can be recovered
     /// given knowledge of the schema of `st_table` and `st_column`.
-    pub tables: BTreeMap<TableId, Vec<Box<Page>>>,
+    pub tables: BTreeMap<TableId, TablePages>,
     /// If the snapshot was compressed or not.
     pub compress_type: CompressType,
     /// Metrics collected while reading this snapshot from disk.