implement compact_filter for hash

Signed-off-by: kevin <kevin@kevin-desktop.lightspeed.mssnks.sbcglobal.net>
Signed-off-by: dashjay <dashjwz@gmail.com>

Author: dashjay

src/storage/compact_filter.cc

@@ -30,6 +30,7 @@
 #include "storage/redis_metadata.h"
 #include "time_util.h"
 #include "types/redis_bitmap.h"
+#include "types/redis_hash.h"
 #include "types/redis_timeseries.h"
 
 namespace engine {
@@ -152,6 +153,48 @@ bool SubKeyFilter::Filter([[maybe_unused]] int level, const Slice &key, const Sl
       return false;
     }
     return expired;
+  } else if (metadata.Type() == kRedisHash) {
+    auto md_expired = IsMetadataExpired(ikey, metadata);
+    if (md_expired) {
+      return true;
+    }
+
+    auto db = stor_->GetDB();
+
+    // Hash field subkeys come in pairs:
+    // 1. Field value key with version = metadata.version
+    // 2. Field expire key with version = metadata.version + ExpireVersionOffset (8)
+    // They are processed independently in compaction, so we must handle both in Filter to ensure consistency.
+    // To avoid orphaning, we check expire time when processing key, but compaction filter handle key one by one, so we
+    // have to left the orphaned key exists here.
+    if (ikey.GetVersion() == metadata.version) {  // when there is an hash field
+      std::string ns_key = ComposeNamespaceKey(ikey.GetNamespace(), ikey.GetKey(), stor_->IsSlotIdEncoded());
+      InternalKey expire_ikey(ns_key, ikey.GetSubKey(), metadata.version + ExpireVersionOffset,
+                              stor_->IsSlotIdEncoded());
+      std::string expire_key = expire_ikey.Encode();
+      std::string expire_value;
+      auto expire_status =
+          db->Get(rocksdb::ReadOptions(), stor_->GetCFHandle(ColumnFamilyID::PrimarySubkey), expire_key, &expire_value);
+      if (expire_status.ok() && expire_value.size() >= sizeof(uint64_t)) {
+        uint64_t expire_at = DecodeFixed64(expire_value.data());
+        uint64_t now = util::GetTimeStampMS();
+        if (expire_at < now) {
+          return true;
+        }
+      }
+      return false;
+    } else if (ikey.GetVersion() == metadata.version + ExpireVersionOffset) {  // when there is an hash field expire key
+      std::string ns_key = ComposeNamespaceKey(ikey.GetNamespace(), ikey.GetKey(), stor_->IsSlotIdEncoded());
+      InternalKey origin_ikey(ns_key, ikey.GetSubKey(), metadata.version, stor_->IsSlotIdEncoded());
+      std::string origin_key = origin_ikey.Encode();
+      std::string origin_value;
+      auto origin_status =
+          db->Get(rocksdb::ReadOptions(), stor_->GetCFHandle(ColumnFamilyID::PrimarySubkey), origin_key, &origin_value);
+      if (origin_status.IsNotFound()) {
+        return true;
+      }
+    }
+    return false;
   }
 
   return IsMetadataExpired(ikey, metadata) || (metadata.Type() == kRedisBitmap && redis::Bitmap::IsEmptySegment(value));

tests/gocase/unit/type/hash/hash_test.go

@@ -1543,6 +1543,24 @@ var testHash = func(t *testing.T, configs util.KvrocksServerConfigs) {
 			result = rdb.Do(ctx, "HEXPIRE", testKey, "60", "FIELDS", "3", "field1", "field2")
 			require.Error(t, result.Err())
 		})
+
+		t.Run("HASH COMPACT", func(t *testing.T) {
+			testKey := "hash-compact-test"
+			require.NoError(t, rdb.Del(ctx, testKey).Err())
+			require.NoError(t, rdb.HSet(ctx, testKey, "field1", "value1").Err())
+
+			// Set expiration to 300 ms
+			rdb.Do(ctx, "HPEXPIRE", testKey, 300, "FIELDS", "1", "field1")
+
+			// Wait for expiration
+			time.Sleep(350 * time.Millisecond)
+
+			// Field should be expired
+			val := rdb.HGet(ctx, testKey, "field1")
+			require.Error(t, val.Err())
+
+			require.NoError(t, rdb.Do(ctx, "COMPACT").Err())
+		})
 	}
 }