fix: fallback to normal delete when deleting a small ranges (#644)

### Motivation

In the current implementation, we will get terrible performance for
iterator-related operations when we have many small `DeleteRange`
ranges.
FYI: https://rocksdb.org/blog/2018/11/21/delete-range.html

It will make our `putWithSequence` method execute very slowly. (based on
the current memtable size)

The reason is that the implementation of pebble(rocksDB) will create the
fragmented range deletion tombstones for range deletion. Even Pebble
will cache the tombstones, but it will invalidate when you get a new
delete range operation. Therefore, we need to rebuild the deletion
iterator after the new `DeleteRange` operation. And it will get worse
along with more `DeleteRange` operations.
FYI:
https://github.com/cockroachdb/pebble/blob/dbc1c128682f7efcdb76352432249780b12447f7/mem_table.go#L248-L251

Also, rocksdb
[article](https://rocksdb.org/blog/2018/11/21/delete-range.html) also
mentioned that in `Future Work` section.

### Modification

- To support atomic operation for the user side, we can fallback the
`DeleteRange` operation back to the normal delete if the actual keys are
lower than `DeleteRangeThreshold(default 100)`.

### Next

1. Introduce policies to make `DeleteRangeThreshold` configurable. 
2. Introduce policies to configure the memtable size. because the
smaller size can trigger flush and compact the deletion tombstones.
3. other small improvements.

Author: mattisonchao

server/kv/db.go

@@ -593,17 +593,41 @@ func (d *db) applyDelete(batch WriteBatch, notifications *notifications, delReq
 	}
 }
 
+const DeleteRangeThreshold = 100
+
 func (d *db) applyDeleteRange(batch WriteBatch, notifications *notifications, delReq *proto.DeleteRangeRequest, updateOperationCallback UpdateOperationCallback) (*proto.DeleteRangeResponse, error) {
 	if notifications != nil {
 		notifications.DeletedRange(delReq.StartInclusive, delReq.EndExclusive)
 	}
 
-	if err := updateOperationCallback.OnDeleteRange(batch, delReq.StartInclusive, delReq.EndExclusive); err != nil {
+	it, err := batch.RangeScan(delReq.StartInclusive, delReq.EndExclusive)
+	if err != nil {
 		return nil, err
 	}
-
-	if err := batch.DeleteRange(delReq.StartInclusive, delReq.EndExclusive); err != nil {
-		return nil, errors.Wrap(err, "oxia db: failed to delete range")
+	var validKeys []string
+	var validKeysNum = 0
+	for ; it.Valid(); it.Next() {
+		validKeysNum++
+		if validKeysNum <= DeleteRangeThreshold {
+			validKeys = append(validKeys, it.Key())
+		}
+		if err = updateOperationCallback.OnDelete(batch, it.Key()); err != nil {
+			return nil, errors.Wrap(multierr.Combine(err, it.Close()), "oxia db: failed to callback on delete range")
+		}
+	}
+	if err := it.Close(); err != nil {
+		return nil, errors.Wrap(err, "oxia db: failed to close iterator on delete range")
+	}
+	if validKeysNum > DeleteRangeThreshold {
+		if err := batch.DeleteRange(delReq.StartInclusive, delReq.EndExclusive); err != nil {
+			return nil, errors.Wrap(err, "oxia db: failed to delete range")
+		}
+	} else {
+		for _, key := range validKeys {
+			if err := batch.Delete(key); err != nil {
+				return nil, errors.Wrap(err, "oxia db: failed to delete range")
+			}
+		}
 	}
 
 	d.log.Debug(

server/kv/db_benchmark_deleterange_test.go

@@ -0,0 +1,57 @@
+// Copyright 2025 StreamNative, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package kv
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"k8s.io/utils/ptr"
+
+	"github.com/streamnative/oxia/common"
+	"github.com/streamnative/oxia/proto"
+)
+
+func BenchmarkDeleteRange(b *testing.B) {
+	factory, err := NewPebbleKVFactory(&FactoryOptions{
+		InMemory:    true,
+		CacheSizeMB: 1024,
+	})
+	assert.NoError(b, err)
+	db, err := NewDB(common.DefaultNamespace, 1, factory, 0, common.SystemClock)
+	assert.NoError(b, err)
+	defer db.Close()
+	for i := range b.N {
+		_, err := db.ProcessWrite(&proto.WriteRequest{
+			Puts: []*proto.PutRequest{
+				{
+					Key:              "00000000000000000001",
+					PartitionKey:     ptr.To("00000000000000000001"),
+					Value:            []byte("00000000000000000000"),
+					SequenceKeyDelta: []uint64{1},
+				},
+			},
+			DeleteRanges: []*proto.DeleteRangeRequest{
+				{
+					StartInclusive: "00000000000000000001-00000000000000000000",
+					EndExclusive:   fmt.Sprintf("00000000000000000001-%020d", i),
+				},
+			},
+		}, int64(i), uint64(time.Now().UnixMilli()), NoOpCallback)
+		assert.NoError(b, err)
+	}
+}