Support batching delete

[niimima]

The following pull request introduces an interface to support IMemoryDbBatchUpsert in batch operations:
#441

I believe it would be even more useful if a batch delete functionality could also be added.

[dluc]

we're planning to migrate all DB operations to run on top of Semantic Kernel vector store connectors. If SK will provide batch delete, we'll probably add that too. Otherwise it will look like a for loop, which could be run similarly on the client side.

[niimima]

By the way, would it not be acceptable to add the current implementation in a manner similar to IMemoryDbBatchUpsert?

From reviewing the source code, I believe it would be possible to implement this without causing any regressions by following the steps below:

• Add the IMemoryDbBatchDelete interface:

namespace Microsoft.KernelMemory.MemoryStorage;

public interface IMemoryDbBatchDelete
{
    public Task DeleteAsync(string index, IEnumerable<MemoryRecord> records, CancellationToken cancellationToken = default);
}

• Modify DeleteDocumentHandler to branch logic based on the interface:

namespace Microsoft.KernelMemory.Handlers;

public sealed class DeleteDocumentHandler : IPipelineStepHandler
{
    ...

    /// <inheritdoc />
    public async Task<(ReturnType returnType, DataPipeline updatedPipeline)> InvokeAsync(
        DataPipeline pipeline, CancellationToken cancellationToken = default)
    {
        this._log.LogDebug("Deleting document, pipeline '{0}/{1}'", pipeline.Index, pipeline.DocumentId);

        // Delete embeddings
        foreach (IMemoryDb db in this._memoryDbs)
        {
            IAsyncEnumerable<MemoryRecord> records = db.GetListAsync(
                index: pipeline.Index,
                limit: -1,
                filters: [MemoryFilters.ByDocument(pipeline.DocumentId)],
                cancellationToken: cancellationToken);

            if (db is IMemoryDbBatchDelete batchDelete)
            {
                // Implement logic to delete records in batches
            }
            else
            {
                await foreach (var record in records.WithCancellation(cancellationToken).ConfigureAwait(false))
                {
                    await db.DeleteAsync(index: pipeline.Index, record, cancellationToken: cancellationToken).ConfigureAwait(false);
                }
            }
        }

        // Delete files, leaving the status file
        await this._documentStorage.EmptyDocumentDirectoryAsync(
            index: pipeline.Index,
            documentId: pipeline.DocumentId,
            cancellationToken).ConfigureAwait(false);

        return (ReturnType.Success, pipeline);
    }
}

[dluc]

Technically yes, but at the cost of increasing the code complexity. When tracking a job progress in the pipeline, it's necessary to checkpoint as often as possible, so that in case of errors the job doesn't restart from the beginning of the list. It's not just a matter of calling one method or the other, the checkpointing logic would be different too, and the cost of maintaining the code.