kvstore: add batch delete (#112723)

-add batch delete to the grpc

pkg/storage/unified/testing/kv.go

@@ -27,6 +27,7 @@ const (
 	TestKVConcurrent     = "concurrent operations"
 	TestKVUnixTimestamp  = "unix timestamp"
 	TestKVBatchGet       = "batch get operations"
+	TestKVBatchDelete    = "batch delete operations"
 )
 
 // NewKVFunc is a function that creates a new KV instance for testing
@@ -67,6 +68,7 @@ func RunKVTest(t *testing.T, newKV NewKVFunc, opts *KVTestOptions) {
 		{TestKVConcurrent, runTestKVConcurrent},
 		{TestKVUnixTimestamp, runTestKVUnixTimestamp},
 		{TestKVBatchGet, runTestKVBatchGet},
+		{TestKVBatchDelete, runTestKVBatchDelete},
 	}
 
 	for _, tc := range cases {
@@ -673,6 +675,122 @@ func runTestKVBatchGet(t *testing.T, kv resource.KV, nsPrefix string) {
 	})
 }
 
+func runTestKVBatchDelete(t *testing.T, kv resource.KV, nsPrefix string) {
+	ctx := testutil.NewTestContext(t, time.Now().Add(30*time.Second))
+	section := nsPrefix + "-batchdelete"
+
+	t.Run("batch delete existing keys", func(t *testing.T) {
+		// Setup test data
+		testData := map[string]string{
+			"key1": "value1",
+			"key2": "value2",
+			"key3": "value3",
+		}
+
+		// Save test data
+		for key, value := range testData {
+			saveKVHelper(t, kv, ctx, section, key, strings.NewReader(value))
+		}
+
+		// Verify keys exist before deletion
+		for key := range testData {
+			_, err := kv.Get(ctx, section, key)
+			require.NoError(t, err)
+		}
+
+		// Batch delete all keys
+		keys := []string{"key1", "key2", "key3"}
+		err := kv.BatchDelete(ctx, section, keys)
+		require.NoError(t, err)
+
+		// Verify all keys are deleted
+		for _, key := range keys {
+			_, err := kv.Get(ctx, section, key)
+			assert.Error(t, err)
+			assert.Equal(t, resource.ErrNotFound, err)
+		}
+	})
+
+	t.Run("batch delete with non-existent keys", func(t *testing.T) {
+		// Setup some test data
+		saveKVHelper(t, kv, ctx, section, "existing-key-1", strings.NewReader("value1"))
+		saveKVHelper(t, kv, ctx, section, "existing-key-2", strings.NewReader("value2"))
+
+		// Batch delete with mix of existing and non-existent keys
+		keys := []string{"existing-key-1", "non-existent-1", "existing-key-2", "non-existent-2"}
+		err := kv.BatchDelete(ctx, section, keys)
+		require.NoError(t, err)
+
+		// Verify existing keys are deleted
+		_, err = kv.Get(ctx, section, "existing-key-1")
+		assert.Error(t, err)
+		assert.Equal(t, resource.ErrNotFound, err)
+
+		_, err = kv.Get(ctx, section, "existing-key-2")
+		assert.Error(t, err)
+		assert.Equal(t, resource.ErrNotFound, err)
+	})
+
+	t.Run("batch delete with all non-existent keys", func(t *testing.T) {
+		// Batch delete keys that don't exist
+		keys := []string{"non-existent-1", "non-existent-2", "non-existent-3"}
+		err := kv.BatchDelete(ctx, section, keys)
+		require.NoError(t, err)
+	})
+
+	t.Run("batch delete with empty keys list", func(t *testing.T) {
+		keys := []string{}
+		err := kv.BatchDelete(ctx, section, keys)
+		require.NoError(t, err)
+	})
+
+	t.Run("batch delete with empty section", func(t *testing.T) {
+		keys := []string{"some-key"}
+		err := kv.BatchDelete(ctx, "", keys)
+		assert.Error(t, err)
+		assert.Contains(t, err.Error(), "section is required")
+	})
+
+	t.Run("batch delete preserves other keys", func(t *testing.T) {
+		// Setup test data
+		saveKVHelper(t, kv, ctx, section, "keep-key-1", strings.NewReader("keep-value-1"))
+		saveKVHelper(t, kv, ctx, section, "delete-key-1", strings.NewReader("delete-value-1"))
+		saveKVHelper(t, kv, ctx, section, "keep-key-2", strings.NewReader("keep-value-2"))
+		saveKVHelper(t, kv, ctx, section, "delete-key-2", strings.NewReader("delete-value-2"))
+
+		// Batch delete specific keys
+		keys := []string{"delete-key-1", "delete-key-2"}
+		err := kv.BatchDelete(ctx, section, keys)
+		require.NoError(t, err)
+
+		// Verify deleted keys are gone
+		_, err = kv.Get(ctx, section, "delete-key-1")
+		assert.Error(t, err)
+		assert.Equal(t, resource.ErrNotFound, err)
+
+		_, err = kv.Get(ctx, section, "delete-key-2")
+		assert.Error(t, err)
+		assert.Equal(t, resource.ErrNotFound, err)
+
+		// Verify kept keys still exist
+		reader, err := kv.Get(ctx, section, "keep-key-1")
+		require.NoError(t, err)
+		value, err := io.ReadAll(reader)
+		require.NoError(t, err)
+		assert.Equal(t, "keep-value-1", string(value))
+		err = reader.Close()
+		require.NoError(t, err)
+
+		reader, err = kv.Get(ctx, section, "keep-key-2")
+		require.NoError(t, err)
+		value, err = io.ReadAll(reader)
+		require.NoError(t, err)
+		assert.Equal(t, "keep-value-2", string(value))
+		err = reader.Close()
+		require.NoError(t, err)
+	})
+}
+
 // saveKVHelper is a helper function to save data to KV store using the new WriteCloser interface
 func saveKVHelper(t *testing.T, kv resource.KV, ctx context.Context, section, key string, value io.Reader) {
 	t.Helper()