Added a test for the case of adding new columns into the existing Roc…

…ksDB database (#1853)

Closes #162

Added a test case to verify the database's behavior when new columns are
added to the RocksDB database.

### Before requesting review
- [x] I have reviewed the code myself

---------

Co-authored-by: Simran Kedia <simrankedia@users.noreply.github.com>

CHANGELOG.md

@@ -14,9 +14,11 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 - [#1866](https://github.com/FuelLabs/fuel-core/pull/1866): Fixed a runtime panic that occurred when restarting a node. The panic happens when the relayer database is already populated, and the relayer attempts an empty commit during start up. This invalid commit is removed in this PR.
 - [#1871](https://github.com/FuelLabs/fuel-core/pull/1871): Fixed `block` endpoint to return fetch the blocks from both databases after regenesis.
 - [#1856](https://github.com/FuelLabs/fuel-core/pull/1856): Replaced instances of `Union` with `Enum` for GraphQL definitions of `ConsensusParametersVersion` and related types. This is needed because `Union` does not support multiple `Version`s inside discriminants or empty variants. 
+- [#1851](https://github.com/FuelLabs/fuel-core/pull/1851/): Provided migration capabilities (enabled addition of new column families) to RocksDB instance.
 
 ### Added 
 
+- [#1853](https://github.com/FuelLabs/fuel-core/pull/1853): Added a test case to verify the database's behavior when new columns are added to the RocksDB database.
 - [#1860](https://github.com/FuelLabs/fuel-core/pull/1860): Regenesis now preserves `FuelBlockIdsToHeights` off-chain table.
 
 ### Changed

crates/database/Cargo.toml

@@ -24,3 +24,6 @@ fuel-core-types = { workspace = true }
 
 [dev-dependencies]
 fuel-core-trace = { path = "../trace" }
+
+[features]
+test-helpers = []

crates/database/src/lib.rs

@@ -64,6 +64,13 @@ pub enum Error {
     Other(anyhow::Error),
 }
 
+#[cfg(feature = "test-helpers")]
+impl PartialEq for Error {
+    fn eq(&self, other: &Self) -> bool {
+        self.to_string().eq(&other.to_string())
+    }
+}
+
 impl From<Error> for anyhow::Error {
     fn from(error: Error) -> Self {
         anyhow::Error::msg(error)

crates/fuel-core/Cargo.toml

@@ -82,6 +82,7 @@ p2p = ["dep:fuel-core-p2p", "dep:fuel-core-sync"]
 relayer = ["dep:fuel-core-relayer"]
 rocksdb = ["dep:rocksdb", "dep:tempfile", "dep:num_cpus"]
 test-helpers = [
+  "fuel-core-database/test-helpers",
   "fuel-core-p2p?/test-helpers",
   "fuel-core-storage/test-helpers",
   "fuel-core-chain-config/test-helpers",

crates/fuel-core/src/state/rocks_db.rs

@@ -202,13 +202,6 @@ where
         }
         block_opts.set_bloom_filter(10.0, true);
 
-        let cf_descriptors = columns.clone().into_iter().map(|i| {
-            ColumnFamilyDescriptor::new(
-                Self::col_name(i.id()),
-                Self::cf_opts(i, &block_opts),
-            )
-        });
-
         let mut opts = Options::default();
         opts.create_if_missing(true);
         opts.set_compression_type(DBCompressionType::Lz4);
@@ -227,36 +220,56 @@ where
             opts.set_row_cache(&cache);
         }
 
-        let db = match DB::open_cf_descriptors(&opts, &path, cf_descriptors) {
-            Err(_) => {
-                // setup cfs
-                match DB::open_cf(&opts, &path, &[] as &[&str]) {
-                    Ok(db) => {
-                        for i in columns {
-                            let opts = Self::cf_opts(i, &block_opts);
-                            db.create_cf(Self::col_name(i.id()), &opts)
-                                .map_err(|e| DatabaseError::Other(e.into()))?;
-                        }
-                        Ok(db)
-                    }
-                    Err(err) => {
-                        tracing::error!("Couldn't open the database with an error: {}. \nTrying to repair the database", err);
-                        DB::repair(&opts, &path)
-                            .map_err(|e| DatabaseError::Other(e.into()))?;
-
-                        let cf_descriptors = columns.clone().into_iter().map(|i| {
-                            ColumnFamilyDescriptor::new(
-                                Self::col_name(i.id()),
-                                Self::cf_opts(i, &block_opts),
-                            )
-                        });
-                        DB::open_cf_descriptors(&opts, &path, cf_descriptors)
-                    }
-                }
+        let existing_column_families = DB::list_cf(&opts, &path).unwrap_or_else(|err| {
+            tracing::error!(
+                "Couldn't get the list of cfs: {}. Returning an empty list",
+                err
+            );
+            vec![]
+        });
+
+        let mut cf_descriptors_to_open = vec![];
+        let mut cf_descriptors_to_create = vec![];
+        for column in columns.clone() {
+            let column_name = Self::col_name(column.id());
+            let opts = Self::cf_opts(column, &block_opts);
+            if existing_column_families.contains(&column_name) {
+                cf_descriptors_to_open.push((column_name, opts));
+            } else {
+                cf_descriptors_to_create.push((column_name, opts));
             }
-            ok => ok,
+        }
+
+        let iterator = cf_descriptors_to_open
+            .clone()
+            .into_iter()
+            .map(|(name, opts)| ColumnFamilyDescriptor::new(name, opts));
+
+        let db = match DB::open_cf_descriptors(&opts, &path, iterator) {
+            Ok(db) => {
+                Ok(db)
+            },
+            Err(err) => {
+                tracing::error!("Couldn't open the database with an error: {}. \nTrying to repair the database", err);
+                DB::repair(&opts, &path)
+                    .map_err(|e| DatabaseError::Other(e.into()))?;
+
+                let iterator = cf_descriptors_to_open
+                    .clone()
+                    .into_iter()
+                    .map(|(name, opts)| ColumnFamilyDescriptor::new(name, opts));
+
+                DB::open_cf_descriptors(&opts, &path, iterator)
+            },
         }
         .map_err(|e| DatabaseError::Other(e.into()))?;
+
+        // Setup cfs
+        for (name, opt) in cf_descriptors_to_create {
+            db.create_cf(name, &opt)
+                .map_err(|e| DatabaseError::Other(e.into()))?;
+        }
+
         let rocks_db = RocksDb {
             db,
             _drop: Default::default(),
@@ -618,6 +631,29 @@ mod tests {
         )
     }
 
+    #[test]
+    fn open_new_columns() {
+        let tmp_dir = TempDir::new().unwrap();
+
+        // Given
+        let old_columns =
+            vec![Column::Coins, Column::Messages, Column::UploadedBytecodes];
+        let database_with_old_columns =
+            RocksDb::<OnChain>::open(tmp_dir.path(), old_columns.clone(), None)
+                .expect("Failed to open database with old columns");
+        drop(database_with_old_columns);
+
+        // When
+        let mut new_columns = old_columns;
+        new_columns.push(Column::ContractsAssets);
+        new_columns.push(Column::Metadata);
+        let database_with_new_columns =
+            RocksDb::<OnChain>::open(tmp_dir.path(), new_columns, None).map(|_| ());
+
+        // Then
+        assert_eq!(Ok(()), database_with_new_columns);
+    }
+
     #[test]
     fn can_put_and_read() {
         let key = vec![0xA, 0xB, 0xC];