kvdb-rocksdb: optional per-column storage path

kvdb-rocksdb/src/lib.rs

@@ -17,7 +17,7 @@
 };
 
 use rocksdb::{
-	BlockBasedOptions, ColumnFamily, ColumnFamilyDescriptor, CompactOptions, Options, ReadOptions, WriteBatch,
+	BlockBasedOptions, ColumnFamily, ColumnFamilyDescriptor, CompactOptions, DBPath, Options, ReadOptions, WriteBatch,
 	WriteOptions, DB,
 };
 
@@ -161,6 +161,13 @@
 	/// orders keys *exactly* the same as the comparator provided to
 	/// previous open calls on the same DB.
 	pub comparator: Option<fn(&[u8], &[u8]) -> cmp::Ordering>,
+
+	/// Directory for this column family's SST files. When set, the column's data is placed on
+	/// this path instead of the main database directory, so a large cold column can live on a
+	/// separate (cheaper) volume. `None` keeps it in the main directory.
+	///
+	/// Only affects where *new* SST files are written; it does not move data already on disk.
+	pub path: Option<PathBuf>,
 }
 
 /// Database configuration
@@ -248,7 +255,7 @@
 	}
 
 	// Get column family configuration with the given block based options.
-	fn column_config(&self, block_opts: &BlockBasedOptions, col: u32) -> Options {
+	fn column_config(&self, block_opts: &BlockBasedOptions, col: u32) -> io::Result<Options> {
 		let column_mem_budget = self.memory_budget_for_col(col);
 		let mut opts = Options::default();
 
@@ -260,8 +267,14 @@
 		if let Some(comparator) = self.columns[col as usize].comparator {
 			opts.set_comparator(&format!("column_{col}_comparator"), Box::new(comparator));
 		}
+		// Place this column family's SST files on a dedicated directory when configured. A single
+		// path with a huge target size pins the whole column there: RocksDB never spills elsewhere,
+		// and `cf_paths` overrides `db_paths` for this column only.
+		if let Some(path) = &self.columns[col as usize].path {
+			opts.set_cf_paths(&[DBPath::new(path, u64::MAX).map_err(other_io_err)?]);
+		}
 
-		opts
+		Ok(opts)
 	}
 }
 
@@ -404,9 +417,11 @@
 		column_names: &[&str],
 		block_opts: &BlockBasedOptions,
 	) -> io::Result<rocksdb::DB> {
-		let cf_descriptors: Vec<_> = (0..config.columns.len())
-			.map(|i| ColumnFamilyDescriptor::new(column_names[i as usize], config.column_config(&block_opts, i as u32)))
-			.collect();
+		let cf_descriptors = (0..config.columns.len())
+			.map(|i| {
+				Ok(ColumnFamilyDescriptor::new(column_names[i as usize], config.column_config(block_opts, i as u32)?))
+			})
+			.collect::<io::Result<Vec<_>>>()?;
 
 		let db = match DB::open_cf_descriptors(&opts, path.as_ref(), cf_descriptors) {
 			Err(_) => {
@@ -415,7 +430,7 @@
 					Ok(mut db) => {
 						for (i, name) in column_names.iter().enumerate() {
 							let _ = db
-								.create_cf(name, &config.column_config(&block_opts, i as u32))
+								.create_cf(name, &config.column_config(block_opts, i as u32)?)
 								.map_err(other_io_err)?;
 						}
 						Ok(db)
@@ -582,7 +597,7 @@
 		let col = column_names.len() as u32;
 		let name = format!("col{}", col);
 		self.config.columns.push(cfg);
-		let col_config = self.config.column_config(&self.block_opts, col as u32);
+		let col_config = self.config.column_config(&self.block_opts, col as u32)?;
 		let _ = db.create_cf(&name, &col_config).map_err(other_io_err)?;
 		column_names.push(name);
 		Ok(())

kvdb-rocksdb/tests/column_path.rs

@@ -0,0 +1,119 @@
+// Copyright 2026 Parity Technologies
+//
+// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
+// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
+// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
+// option. This file may not be copied, modified, or distributed
+// except according to those terms.
+
+//! End-to-end checks for `ColumnConfig::path`: SST files must land in the
+//! override directory, survive reopen, and a reopen without the override must
+//! not silently serve an empty column.
+
+use kvdb::KeyValueDB;
+use kvdb_rocksdb::{Database, DatabaseConfig};
+use std::{fs, path::Path};
+
+const COLD_COL: u32 = 1;
+const NUM_KEYS: u32 = 512;
+const VALUE_LEN: usize = 8 * 1024; // 512 * 8 KiB = 4 MiB, enough to force SST flushes at a 1 MiB budget
+
+fn sst_files(dir: &Path) -> Vec<String> {
+	let mut out = Vec::new();
+	let mut stack = vec![dir.to_path_buf()];
+	while let Some(d) = stack.pop() {
+		if let Ok(entries) = fs::read_dir(&d) {
+			for e in entries.flatten() {
+				let p = e.path();
+				if p.is_dir() {
+					stack.push(p);
+				} else if p.extension().map_or(false, |x| x == "sst") {
+					out.push(p.file_name().unwrap().to_string_lossy().into_owned());
+				}
+			}
+		}
+	}
+	out
+}
+
+fn config(cold_path: Option<&Path>) -> DatabaseConfig {
+	let mut cfg = DatabaseConfig::with_columns(2);
+	cfg.columns[COLD_COL as usize].memory_budget = Some(1); // tiny budget so 4 MiB of writes flushes to SSTs
+	cfg.columns[COLD_COL as usize].path = cold_path.map(|p| p.to_path_buf());
+	cfg
+}
+
+fn value(i: u32) -> Vec<u8> {
+	let mut v = vec![0u8; VALUE_LEN];
+	v[..4].copy_from_slice(&i.to_le_bytes());
+	v
+}
+
+fn write_all(db: &Database) {
+	for i in 0..NUM_KEYS {
+		let mut tx = db.transaction();
+		tx.put(COLD_COL, &i.to_le_bytes(), &value(i));
+		if i == 0 {
+			tx.put(0, b"main", b"stays-on-main-path");
+		}
+		db.write(tx).unwrap();
+	}
+}
+
+fn read_all(db: &Database) {
+	for i in 0..NUM_KEYS {
+		let got = db.get(COLD_COL, &i.to_le_bytes()).unwrap();
+		assert_eq!(got.as_deref(), Some(value(i).as_slice()), "key {} lost", i);
+	}
+	assert_eq!(db.get(0, b"main").unwrap().as_deref(), Some(&b"stays-on-main-path"[..]));
+}
+
+#[test]
+fn cold_column_ssts_land_on_override_path_and_survive_reopen() {
+	let main = tempfile::tempdir().unwrap();
+	let cold = tempfile::tempdir().unwrap();
+	let cold_dir = cold.path().join("col1");
+
+	let db = Database::open(&config(Some(&cold_dir)), main.path()).unwrap();
+	write_all(&db);
+	read_all(&db);
+	drop(db);
+
+	let cold_ssts = sst_files(&cold_dir);
+	assert!(!cold_ssts.is_empty(), "no SST files under the override dir {:?}", cold_dir);
+	let main_ssts = sst_files(main.path());
+	for f in &cold_ssts {
+		assert!(!main_ssts.contains(f), "SST {} present in BOTH main and override dirs", f);
+	}
+
+	let db = Database::open(&config(Some(&cold_dir)), main.path()).unwrap();
+	read_all(&db);
+}
+
+#[test]
+fn reopen_without_override_does_not_silently_serve_data() {
+	let main = tempfile::tempdir().unwrap();
+	let cold = tempfile::tempdir().unwrap();
+	let cold_dir = cold.path().join("col1");
+
+	let db = Database::open(&config(Some(&cold_dir)), main.path()).unwrap();
+	write_all(&db);
+	drop(db);
+	assert!(!sst_files(&cold_dir).is_empty());
+
+	// Simulates a node restarted with the flag dropped (or the volume unmounted): RocksDB must
+	// refuse to open rather than come up with a hole where the cold column was.
+	match Database::open(&config(None), main.path()) {
+		Err(e) => {
+			eprintln!("reopen without override failed loudly (good): {}", e);
+		},
+		Ok(db) => {
+			let sample = db.get(COLD_COL, &0u32.to_le_bytes()).unwrap();
+			assert!(
+				sample.is_none(),
+				"reopen without override silently served cold-column data; placement suspect"
+			);
+			panic!("reopen without override succeeded with an empty cold column: silent data hole");
+		},
+	}
+}