diff --git a/pkg/storage/unified/resource/data/sqlkv_delete.sql b/pkg/storage/unified/resource/data/sqlkv_delete.sql
new file mode 100644
index 00000000000..60a60746993
--- /dev/null
+++ b/pkg/storage/unified/resource/data/sqlkv_delete.sql
@@ -0,0 +1,3 @@
+DELETE
+FROM {{ .Ident .TableName }}
+WHERE {{ .Ident "key_path" }} = {{ .Arg .KeyPath }};
diff --git a/pkg/storage/unified/resource/kv.go b/pkg/storage/unified/resource/kv.go
index 0af2cc3c4a1..0bd413254a4 100644
--- a/pkg/storage/unified/resource/kv.go
+++ b/pkg/storage/unified/resource/kv.go
@@ -228,6 +228,9 @@ func (k *badgerKV) Delete(ctx context.Context, section string, key string) error
 	if section == "" {
 		return fmt.Errorf("section is required")
 	}
+	if key == "" {
+		return fmt.Errorf("key is required")
+	}
 
 	txn := k.db.NewTransaction(true)
 	defer txn.Discard()
diff --git a/pkg/storage/unified/resource/sqlkv.go b/pkg/storage/unified/resource/sqlkv.go
index c943b5292f6..90171a77a5f 100644
--- a/pkg/storage/unified/resource/sqlkv.go
+++ b/pkg/storage/unified/resource/sqlkv.go
@@ -33,9 +33,13 @@ func mustTemplate(filename string) *template.Template {
 
 // Templates.
 var (
-	sqlKVGet = mustTemplate("sqlkv_get.sql")
+	sqlKVGet    = mustTemplate("sqlkv_get.sql")
+	sqlKVDelete = mustTemplate("sqlkv_delete.sql")
 )
 
+// sqlKVSection can be embedded in structs used when rendering query templates
+// for queries that reference a particular section. The section will be validated,
+// and the template can directly reference the `TableName`.
 type sqlKVSection struct {
 	Section string
 }
@@ -60,18 +64,15 @@ func (req sqlKVSection) TableName() string {
 	return "resource_events"
 }
 
-type sqlKVGetRequest struct {
-	sqltemplate.SQLTemplate
+// sqlKVSectionKey can be embedded in structs used when rendering query templates
+// for queries that reference both a section and a particular key. The `key` is
+// validated, and the template can reference the corresponding `KeyPath`.
+type sqlKVSectionKey struct {
 	sqlKVSection
 	Key string
-	*sqlKVGetResponse
 }
 
-type sqlKVGetResponse struct {
-	Value []byte
-}
-
-func (req sqlKVGetRequest) Validate() error {
+func (req sqlKVSectionKey) Validate() error {
 	if err := req.sqlKVSection.Validate(); err != nil {
 		return err
 	}
@@ -82,12 +83,35 @@ func (req sqlKVGetRequest) Validate() error {
 	return nil
 }
 
+func (req sqlKVSectionKey) KeyPath() string {
+	return req.Section + "/" + req.Key
+}
+
+type sqlKVGetRequest struct {
+	sqltemplate.SQLTemplate
+	sqlKVSectionKey
+	*sqlKVGetResponse
+}
+
+type sqlKVGetResponse struct {
+	Value []byte
+}
+
+func (req sqlKVGetRequest) Validate() error {
+	return req.sqlKVSectionKey.Validate()
+}
+
 func (req sqlKVGetRequest) Results() ([]byte, error) {
 	return req.Value, nil
 }
 
-func (req sqlKVGetRequest) KeyPath() string {
-	return req.Section + "/" + req.Key
+type sqlKVDeleteRequest struct {
+	sqltemplate.SQLTemplate
+	sqlKVSectionKey
+}
+
+func (req sqlKVDeleteRequest) Validate() error {
+	return req.sqlKVSectionKey.Validate()
 }
 
 var _ KV = &sqlKV{}
@@ -134,8 +158,7 @@ func (k *sqlKV) Keys(ctx context.Context, section string, opt ListOptions) iter.
 func (k *sqlKV) Get(ctx context.Context, section string, key string) (io.ReadCloser, error) {
 	value, err := dbutil.QueryRow(ctx, k.db, sqlKVGet, sqlKVGetRequest{
 		SQLTemplate:      sqltemplate.New(k.dialect),
-		sqlKVSection:     sqlKVSection{section},
-		Key:              key,
+		sqlKVSectionKey:  sqlKVSectionKey{sqlKVSection{section}, key},
 		sqlKVGetResponse: new(sqlKVGetResponse),
 	})
 	if err != nil {
@@ -171,7 +194,24 @@ func (k *sqlKV) Save(ctx context.Context, section string, key string) (io.WriteC
 }
 
 func (k *sqlKV) Delete(ctx context.Context, section string, key string) error {
-	panic("not implemented!")
+	res, err := dbutil.Exec(ctx, k.db, sqlKVDelete, sqlKVDeleteRequest{
+		SQLTemplate:     sqltemplate.New(k.dialect),
+		sqlKVSectionKey: sqlKVSectionKey{sqlKVSection{section}, key},
+	})
+	if err != nil {
+		return fmt.Errorf("failed to delete key: %w", err)
+	}
+
+	rows, err := res.RowsAffected()
+	if err != nil {
+		return fmt.Errorf("failed to validate delete: %w", err)
+	}
+
+	if rows == 0 {
+		return ErrNotFound
+	}
+
+	return nil
 }
 
 func (k *sqlKV) BatchDelete(ctx context.Context, section string, keys []string) error {
diff --git a/pkg/storage/unified/testing/kv.go b/pkg/storage/unified/testing/kv.go
index 9688c5c1029..fbdb8c70d20 100644
--- a/pkg/storage/unified/testing/kv.go
+++ b/pkg/storage/unified/testing/kv.go
@@ -28,6 +28,10 @@ const (
 	TestKVUnixTimestamp  = "unix timestamp"
 	TestKVBatchGet       = "batch get operations"
 	TestKVBatchDelete    = "batch delete operations"
+
+	// Use `eventsSection` as the section for the tests, as the sqlkv implementation
+	// needs a real section to determine which table to use.
+	testSection = "unified/events"
 )
 
 // NewKVFunc is a function that creates a new KV instance for testing
@@ -84,22 +88,21 @@ func RunKVTest(t *testing.T, newKV NewKVFunc, opts *KVTestOptions) {
 	}
 }
 
+func prefixKey(nsPrefix, key string) string {
+	return nsPrefix + "/" + key
+}
+
 func runTestKVGet(t *testing.T, kv resource.KV, nsPrefix string) {
 	ctx := testutil.NewTestContext(t, time.Now().Add(30*time.Second))
-	// Use `eventsSection` as the section for these tests, as the sqlkv implementation
-	// needs a real section to determine which table to use, and apply `nsPrefix` on
-	// the key itself.
-	section := "unified/events"
-	keyPrefix := nsPrefix + "-get"
-	prefixed := func(name string) string { return keyPrefix + "/" + name }
 
 	t.Run("get existing key", func(t *testing.T) {
 		// First save a key
+		existingKey := prefixKey(nsPrefix, "existing-key")
 		testValue := "test value for get"
-		saveKVHelper(t, kv, ctx, section, prefixed("existing-key"), strings.NewReader(testValue))
+		saveKVHelper(t, kv, ctx, testSection, existingKey, strings.NewReader(testValue))
 
 		// Now get it
-		reader, err := kv.Get(ctx, section, prefixed("existing-key"))
+		reader, err := kv.Get(ctx, testSection, existingKey)
 		require.NoError(t, err)
 
 		// Read the value
@@ -113,19 +116,19 @@ func runTestKVGet(t *testing.T, kv resource.KV, nsPrefix string) {
 	})
 
 	t.Run("get non-existent key", func(t *testing.T) {
-		_, err := kv.Get(ctx, section, prefixed("non-existent-key"))
+		_, err := kv.Get(ctx, testSection, prefixKey(nsPrefix, "non-existent-key"))
 		assert.Error(t, err)
 		assert.Equal(t, resource.ErrNotFound, err)
 	})
 
 	t.Run("get with empty section", func(t *testing.T) {
-		_, err := kv.Get(ctx, "", prefixed("some-key"))
+		_, err := kv.Get(ctx, "", prefixKey(nsPrefix, "some-key"))
 		assert.Error(t, err)
 		assert.Contains(t, err.Error(), "section is required")
 	})
 
 	t.Run("get with empty key", func(t *testing.T) {
-		_, err := kv.Get(ctx, section, "")
+		_, err := kv.Get(ctx, testSection, "")
 		assert.Error(t, err)
 		assert.Contains(t, err.Error(), "key is required")
 	})
@@ -209,37 +212,43 @@ func runTestKVSave(t *testing.T, kv resource.KV, nsPrefix string) {
 
 func runTestKVDelete(t *testing.T, kv resource.KV, nsPrefix string) {
 	ctx := testutil.NewTestContext(t, time.Now().Add(30*time.Second))
-	section := nsPrefix + "-delete"
 
 	t.Run("delete existing key", func(t *testing.T) {
 		// First create a key
-		saveKVHelper(t, kv, ctx, section, "delete-key", strings.NewReader("delete me"))
+		deleteKey := prefixKey(nsPrefix, "delete-key")
+		saveKVHelper(t, kv, ctx, testSection, deleteKey, strings.NewReader("delete me"))
 
 		// Verify it exists
-		_, err := kv.Get(ctx, section, "delete-key")
+		_, err := kv.Get(ctx, testSection, deleteKey)
 		require.NoError(t, err)
 
 		// Delete it
-		err = kv.Delete(ctx, section, "delete-key")
+		err = kv.Delete(ctx, testSection, deleteKey)
 		require.NoError(t, err)
 
 		// Verify it's gone
-		_, err = kv.Get(ctx, section, "delete-key")
+		_, err = kv.Get(ctx, testSection, deleteKey)
 		assert.Error(t, err)
 		assert.Equal(t, resource.ErrNotFound, err)
 	})
 
 	t.Run("delete non-existent key", func(t *testing.T) {
-		err := kv.Delete(ctx, section, "non-existent-delete-key")
+		err := kv.Delete(ctx, testSection, prefixKey(nsPrefix, "non-existent-delete-key"))
 		assert.Error(t, err)
 		assert.Equal(t, resource.ErrNotFound, err)
 	})
 
 	t.Run("delete with empty section", func(t *testing.T) {
-		err := kv.Delete(ctx, "", "some-key")
+		err := kv.Delete(ctx, "", prefixKey(nsPrefix, "some-key"))
 		assert.Error(t, err)
 		assert.Contains(t, err.Error(), "section is required")
 	})
+
+	t.Run("delete with empty key", func(t *testing.T) {
+		err := kv.Delete(ctx, testSection, "")
+		assert.Error(t, err)
+		assert.Contains(t, err.Error(), "key is required")
+	})
 }
 
 func runTestKVKeys(t *testing.T, kv resource.KV, nsPrefix string) {
diff --git a/pkg/storage/unified/testing/kv_test.go b/pkg/storage/unified/testing/kv_test.go
index 4fc343d450e..f6db1d2f1d8 100644
--- a/pkg/storage/unified/testing/kv_test.go
+++ b/pkg/storage/unified/testing/kv_test.go
@@ -48,7 +48,6 @@ func TestSQLKV(t *testing.T) {
 		NSPrefix: "sql-kv-test",
 		SkipTests: map[string]bool{
 			TestKVSave:           true,
-			TestKVDelete:         true,
 			TestKVKeys:           true,
 			TestKVKeysWithLimits: true,
 			TestKVKeysWithSort:   true,