fix: tombstone vector sets on DEL

Author: happy-v587

src/storage/src/redis_strings.rs

@@ -33,7 +33,6 @@ use crate::{
     format_base_meta_value::ParsedBaseMetaValue,
     format_list_meta_value::ParsedListsMetaValue,
     format_strings_value::{ParsedStringsValue, StringValue},
-    format_vector_member_key::VectorMemberDataKey,
     redis_sets::glob_match_bytes,
 };
 
@@ -2059,7 +2058,7 @@ impl Redis {
                     log::warn!("failed to update key statistics for {key_str}: {error:?}");
                 }
             }
-            DataType::Hash | DataType::Set | DataType::ZSet => {
+            DataType::Hash | DataType::Set | DataType::ZSet | DataType::VectorSet => {
                 let mut parsed = ParsedBaseMetaValue::new(&value[..])?;
                 if !parsed.is_valid() {
                     return Ok(false);
@@ -2081,39 +2080,6 @@ impl Redis {
                     log::warn!("failed to update key statistics for {key_str}: {error:?}");
                 }
             }
-            DataType::VectorSet => {
-                let parsed = ParsedBaseMetaValue::new(&value[..])?;
-                if !parsed.is_valid() {
-                    return Ok(false);
-                }
-                let count = parsed.count();
-                let prefix = VectorMemberDataKey::encode_key_prefix(key)?;
-                let Some(vector_cf) = self.get_cf_handle(ColumnFamilyIndex::VectorDataCF) else {
-                    return Ok(false);
-                };
-                let mut keys_to_delete = Vec::new();
-                for item in db.iterator_cf(
-                    &vector_cf,
-                    rocksdb::IteratorMode::From(&prefix, rocksdb::Direction::Forward),
-                ) {
-                    let (member_key, _) = item.context(RocksSnafu)?;
-                    if !member_key.starts_with(&prefix) {
-                        break;
-                    }
-                    keys_to_delete.push(member_key.to_vec());
-                }
-                let mut batch = self.create_batch()?;
-                batch.delete(ColumnFamilyIndex::MetaCF, &encoded_meta_key)?;
-                for member_key in keys_to_delete {
-                    batch.delete(ColumnFamilyIndex::VectorDataCF, &member_key)?;
-                }
-                batch.commit()?;
-                if let Err(error) =
-                    self.update_specific_key_statistics(DataType::VectorSet, &key_str, count)
-                {
-                    log::warn!("failed to update key statistics for {key_str}: {error:?}");
-                }
-            }
             DataType::List => {
                 let mut parsed = ParsedListsMetaValue::new(&value[..])?;
                 if !parsed.is_valid() {

src/storage/tests/redis_vector_test.rs

@@ -653,16 +653,25 @@ async fn test_expired_vectorset_reads_as_missing() {
 }
 
 #[test]
-fn test_del_removes_vector_meta_and_members() {
+fn test_del_tombstones_vector_meta_and_defers_member_cleanup() {
     with_redis(|redis| {
         let vector = CanonicalVector::from_values(&[1.0, 0.0]).expect("vector");
         redis.vadd(b"vectors", b"a", &vector).expect("insert a");
         redis.vadd(b"vectors", b"b", &vector).expect("insert b");
         assert_eq!(count_cf_entries(redis, ColumnFamilyIndex::VectorDataCF), 2);
+        let previous_version = read_stored_vector_meta(redis, b"vectors").version();
 
         assert!(redis.del_key(b"vectors").expect("delete vector set"));
         assert_eq!(redis.vcard(b"vectors").expect("missing card"), 0);
-        assert_eq!(count_cf_entries(redis, ColumnFamilyIndex::VectorDataCF), 0);
+        let tombstone = read_stored_vector_meta(redis, b"vectors");
+        assert_eq!(tombstone.count(), 0);
+        assert_eq!(tombstone.etime(), 0);
+        assert!(tombstone.version() > previous_version);
+        assert_eq!(
+            count_cf_entries(redis, ColumnFamilyIndex::VectorDataCF),
+            2,
+            "DEL should leave vector members for compaction"
+        );
     });
 }