Modify nydus-image chunkdict save

Add two fields  image_name and version_name of the chunk table in the database
Modify the database storage logic to save data without erasing the original data

Signed-off-by: Zhao Yuan <[email protected]>

Author: newthifans

src/bin/nydus-image/deduplicate.rs

@@ -11,6 +11,7 @@ use nydus_storage::device::BlobInfo;
 use rusqlite::{params, Connection};
 use std::fs;
 use std::path::Path;
+use std::result::Result::Ok;
 use std::sync::{Arc, Mutex};
 
 #[derive(Debug)]
@@ -66,15 +67,11 @@ pub struct SqliteDatabase {
 
 impl SqliteDatabase {
     pub fn new(database_url: &str) -> Result<Self, rusqlite::Error> {
-        // Delete the database file if it exists.
+        // Connect to a database that already exists
         if let Ok(metadata) = fs::metadata(database_url) {
             if metadata.is_file() {
-                if let Err(err) = fs::remove_file(database_url) {
-                    warn!(
-                        "Warning: Unable to delete existing database file: {:?}.",
-                        err
-                    );
-                }
+            } else {
+                panic!("Warning: Unable to find existing database file.");
             }
         }
 
@@ -95,24 +92,6 @@ impl SqliteDatabase {
             blob_table,
         })
     }
-
-    pub fn connect(database_url: &str) -> Result<Self, rusqlite::Error> {
-        // Connect to a database that already exists
-        if let Ok(metadata) = fs::metadata(database_url) {
-            if metadata.is_file() {
-            } else {
-                panic!("Warning: Unable to find existing database file.");
-            }
-        }
-
-        let chunk_table = ChunkTable::new(database_url)?;
-        let blob_table = BlobTable::new(database_url)?;
-
-        Ok(Self {
-            chunk_table,
-            blob_table,
-        })
-    }
 }
 
 impl Database for SqliteDatabase {
@@ -163,11 +142,15 @@ impl Deduplicate<SqliteDatabase> {
         Ok(Self { sb, db })
     }
 
-    pub fn save_metadata(&mut self) -> anyhow::Result<Vec<Arc<BlobInfo>>> {
+    pub fn save_metadata(
+        &mut self,
+        image_name: String,
+        version_name: String,
+    ) -> anyhow::Result<Vec<Arc<BlobInfo>>> {
         self.create_tables()?;
         let blob_infos = self.sb.superblock.get_blob_infos();
         self.insert_blobs(&blob_infos)?;
-        self.insert_chunks(&blob_infos)?;
+        self.insert_chunks(&blob_infos, image_name, version_name)?;
         Ok(blob_infos)
     }
 
@@ -194,14 +177,21 @@ impl Deduplicate<SqliteDatabase> {
         Ok(())
     }
 
-    fn insert_chunks(&mut self, blob_infos: &[Arc<BlobInfo>]) -> anyhow::Result<()> {
+    fn insert_chunks(
+        &mut self,
+        blob_infos: &[Arc<BlobInfo>],
+        image_name: String,
+        version_name: String,
+    ) -> anyhow::Result<()> {
         let process_chunk = &mut |t: &Tree| -> Result<()> {
             let node = t.lock_node();
             for chunk in &node.chunks {
                 let index = chunk.inner.blob_index();
                 let chunk_blob_id = blob_infos[index as usize].blob_id();
                 self.db
                     .insert_chunk(&Chunk {
+                        image_name: image_name.to_string(),
+                        version_name: version_name.to_string(),
                         chunk_blob_id,
                         chunk_digest: chunk.inner.id().to_string(),
                         chunk_compressed_size: chunk.inner.compressed_size(),
@@ -220,75 +210,14 @@ impl Deduplicate<SqliteDatabase> {
     }
 }
 
-pub struct Algorithm<D: Database + Send + Sync> {
-    algorithm_name: String,
-    db: D,
-}
-
-impl Algorithm<SqliteDatabase> {
-    pub fn new(algorithm: String, db_url: &str) -> anyhow::Result<Self> {
-        let algorithm_name = algorithm;
-        let db = SqliteDatabase::connect(db_url)?;
-        Ok(Self { algorithm_name, db })
-    }
-
-    pub fn chunkdict_generate(&mut self) -> anyhow::Result<Vec<Chunk>> {
-        let all_chunks = self.db.chunk_table.list_all()?;
-        let chunkdict = match &self.algorithm_name as &str {
-            "exponential_smoothing" => Self::exponential_smoothing(self, all_chunks)?,
-            _ => {
-                bail!("Unsupported algorithm name:, please use a valid algorithm name, such as exponential_smoothing")
-            }
-        };
-        Ok(chunkdict)
-    }
-
-    // Algorithm "exponential_smoothing"
-    // List all chunk and sort them by the order in chunk table
-    // Score each chunk by "exponential_smoothing" formula
-    // Select chunks whose score is greater than threshold and generate chunk dictionary
-    fn exponential_smoothing(&mut self, all_chunks: Vec<Chunk>) -> anyhow::Result<Vec<Chunk>> {
-        let alpha = 0.5;
-        let previou_length = 1000;
-        let threshold = 0.1;
-        let mut smoothed_data = Vec::new();
-        for (chunk_index, chunk) in all_chunks.iter().enumerate() {
-            let mut is_duplicate: f64 = 0.0;
-            let mut temp = 0;
-            if chunk_index > previou_length {
-                temp = chunk_index - previou_length;
-            }
-
-            for previou_index in all_chunks.iter().take(chunk_index).skip(temp) {
-                if chunk.chunk_digest == previou_index.chunk_digest {
-                    is_duplicate = 1.0;
-                    break;
-                }
-            }
-            if chunk_index == 0 {
-                smoothed_data.push(0.0);
-            } else {
-                let smoothed_score: f64 =
-                    alpha * is_duplicate + (1.0 - alpha) * smoothed_data[chunk_index - 1];
-                smoothed_data.push(smoothed_score);
-            }
-        }
-        let mut chunkdict: Vec<Chunk> = Vec::new();
-        for i in 0..smoothed_data.len() {
-            let chunk = Chunk {
-                chunk_blob_id: all_chunks[i].chunk_blob_id.clone(),
-                chunk_digest: all_chunks[i].chunk_digest.clone(),
-                chunk_compressed_offset: all_chunks[i].chunk_compressed_offset,
-                chunk_uncompressed_offset: all_chunks[i].chunk_uncompressed_offset,
-                chunk_compressed_size: all_chunks[i].chunk_compressed_size,
-                chunk_uncompressed_size: all_chunks[i].chunk_uncompressed_size,
-            };
-            if smoothed_data[i] > threshold {
-                chunkdict.push(chunk);
-            }
-        }
-        Ok(chunkdict)
-    }
+#[allow(dead_code)]
+#[derive(Debug)]
+struct DataPoint {
+    image_name: String,
+    chunk_list: Vec<Chunk>,
+    visited: bool,
+    clustered: bool,
+    cluster_id: i32,
 }
 
 pub trait Table<T, Err>: Sync + Send + Sized + 'static
@@ -311,7 +240,7 @@ where
     fn list_paged(&self, offset: i64, limit: i64) -> Result<Vec<T>, Err>;
 }
 
-#[derive(Debug)]
+#[derive()]
 pub struct ChunkTable {
     conn: Arc<Mutex<Connection>>,
 }
@@ -332,8 +261,10 @@ impl ChunkTable {
     }
 }
 
-#[derive(Debug)]
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
 pub struct Chunk {
+    image_name: String,
+    version_name: String,
     chunk_blob_id: String,
     chunk_digest: String,
     chunk_compressed_size: u32,
@@ -359,6 +290,8 @@ impl Table<Chunk, DatabaseError> for ChunkTable {
             .execute(
                 "CREATE TABLE IF NOT EXISTS chunk (
                     id               INTEGER PRIMARY KEY,
+                    image_name       TEXT,
+                    version_name     TEXT,
                     chunk_blob_id    TEXT NOT NULL,
                     chunk_digest     TEXT,
                     chunk_compressed_size  INT,
@@ -378,16 +311,20 @@ impl Table<Chunk, DatabaseError> for ChunkTable {
             .map_err(|e| DatabaseError::PoisonError(e.to_string()))?
             .execute(
                 "INSERT INTO chunk(
+                    image_name,
+                    version_name,
                     chunk_blob_id,
                     chunk_digest,
                     chunk_compressed_size,
                     chunk_uncompressed_size,
                     chunk_compressed_offset,
                     chunk_uncompressed_offset
                 )
-                VALUES (?1, ?2, ?3, ?4, ?5, ?6);
+                VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8);
                 ",
                 rusqlite::params![
+                    chunk.image_name,
+                    chunk.version_name,
                     chunk.chunk_blob_id,
                     chunk.chunk_digest,
                     chunk.chunk_compressed_size,
@@ -425,18 +362,20 @@ impl Table<Chunk, DatabaseError> for ChunkTable {
             .map_err(|e| DatabaseError::PoisonError(e.to_string()))?;
         let mut stmt: rusqlite::Statement<'_> = conn_guard
             .prepare(
-                "SELECT id, chunk_blob_id, chunk_digest, chunk_compressed_size,
+                "SELECT id, image_name, version_name, chunk_blob_id, chunk_digest, chunk_compressed_size,
                 chunk_uncompressed_size, chunk_compressed_offset, chunk_uncompressed_offset from chunk
                 ORDER BY id LIMIT ?1 OFFSET ?2",
             )?;
         let chunk_iterator = stmt.query_map(params![limit, offset], |row| {
             Ok(Chunk {
-                chunk_blob_id: row.get(1)?,
-                chunk_digest: row.get(2)?,
-                chunk_compressed_size: row.get(3)?,
-                chunk_uncompressed_size: row.get(4)?,
-                chunk_compressed_offset: row.get(5)?,
-                chunk_uncompressed_offset: row.get(6)?,
+                image_name: row.get(1)?,
+                version_name: row.get(2)?,
+                chunk_blob_id: row.get(3)?,
+                chunk_digest: row.get(4)?,
+                chunk_compressed_size: row.get(5)?,
+                chunk_uncompressed_size: row.get(6)?,
+                chunk_compressed_offset: row.get(7)?,
+                chunk_uncompressed_offset: row.get(8)?,
             })
         })?;
         let mut chunks = Vec::new();
@@ -593,6 +532,8 @@ mod tests {
         let chunk_table = ChunkTable::new_in_memory()?;
         chunk_table.create()?;
         let chunk = Chunk {
+            image_name: "REDIS".to_string(),
+            version_name: "1.0.0".to_string(),
             chunk_blob_id: "BLOB123".to_string(),
             chunk_digest: "DIGEST123".to_string(),
             chunk_compressed_size: 512,
@@ -602,6 +543,8 @@ mod tests {
         };
         chunk_table.insert(&chunk)?;
         let chunks = chunk_table.list_all()?;
+        assert_eq!(chunks[0].image_name, chunk.image_name);
+        assert_eq!(chunks[0].version_name, chunk.version_name);
         assert_eq!(chunks.len(), 1);
         assert_eq!(chunks[0].chunk_blob_id, chunk.chunk_blob_id);
         assert_eq!(chunks[0].chunk_digest, chunk.chunk_digest);
@@ -648,6 +591,8 @@ mod tests {
         for i in 0..200 {
             let i64 = i as u64;
             let chunk = Chunk {
+                image_name: format!("REDIS{}", i),
+                version_name: format!("1.0.0{}", i),
                 chunk_blob_id: format!("BLOB{}", i),
                 chunk_digest: format!("DIGEST{}", i),
                 chunk_compressed_size: i,
@@ -659,6 +604,8 @@ mod tests {
         }
         let chunks = chunk_table.list_paged(100, 100)?;
         assert_eq!(chunks.len(), 100);
+        assert_eq!(chunks[0].image_name, "REDIS100");
+        assert_eq!(chunks[0].version_name, "1.0.0100");
         assert_eq!(chunks[0].chunk_blob_id, "BLOB100");
         assert_eq!(chunks[0].chunk_digest, "DIGEST100");
         assert_eq!(chunks[0].chunk_compressed_size, 100);
@@ -668,38 +615,4 @@ mod tests {
         Ok(())
     }
 
-    #[test]
-    fn test_alalgorithm_exponential_smoothing() -> Result<(), Box<dyn std::error::Error>> {
-        let algorithm = String::from("exponential_smoothing");
-        let db_url = "./metadata.db";
-        let chunk_table = ChunkTable::new(db_url)?;
-        chunk_table.clear()?;
-        chunk_table.create()?;
-        for i in 0..200 {
-            let i64 = i as u64;
-            let chunk = Chunk {
-                chunk_blob_id: format!("BLOB{}", i),
-                chunk_digest: format!("DIGEST{}", (i + 1) % 2),
-                chunk_compressed_size: i,
-                chunk_uncompressed_size: i * 2,
-                chunk_compressed_offset: i64 * 3,
-                chunk_uncompressed_offset: i64 * 4,
-            };
-            chunk_table.insert(&chunk)?;
-        }
-        let mut algorithm = Algorithm::<SqliteDatabase>::new(algorithm, db_url)?;
-        assert_eq!(
-            algorithm.algorithm_name,
-            "exponential_smoothing".to_string()
-        );
-        let chunkdict = algorithm.chunkdict_generate()?;
-        assert_eq!(chunkdict.len(), 198);
-        assert_eq!(chunkdict[0].chunk_blob_id, "BLOB2");
-        assert_eq!(chunkdict[0].chunk_digest, "DIGEST1");
-        assert_eq!(chunkdict[0].chunk_compressed_size, 2);
-        assert_eq!(chunkdict[0].chunk_uncompressed_size, 4);
-        assert_eq!(chunkdict[0].chunk_compressed_offset, 6);
-        assert_eq!(chunkdict[0].chunk_uncompressed_offset, 8);
-        Ok(())
-    }
 }