diff --git a/src/Libraries/Microsoft.Extensions.DataIngestion/Writers/VectorStoreWriter.cs b/src/Libraries/Microsoft.Extensions.DataIngestion/Writers/VectorStoreWriter.cs index f231ac6535b..5b312836732 100644 --- a/src/Libraries/Microsoft.Extensions.DataIngestion/Writers/VectorStoreWriter.cs +++ b/src/Libraries/Microsoft.Extensions.DataIngestion/Writers/VectorStoreWriter.cs @@ -180,6 +180,7 @@ private async Task> GetPreExistingChunksIdsAsync(Ingestion await foreach (var record in _vectorStoreCollection!.GetAsync( filter: record => (string)record[DocumentIdName]! == document.Identifier, top: MaxTopCount, + options: new() { Skip = keys.Count }, cancellationToken: cancellationToken).ConfigureAwait(false)) { keys.Add(record[KeyName]!); diff --git a/test/Libraries/Microsoft.Extensions.DataIngestion.Tests/Writers/VectorStoreWriterTests.cs b/test/Libraries/Microsoft.Extensions.DataIngestion.Tests/Writers/VectorStoreWriterTests.cs index d64eb990176..0600a00216d 100644 --- a/test/Libraries/Microsoft.Extensions.DataIngestion.Tests/Writers/VectorStoreWriterTests.cs +++ b/test/Libraries/Microsoft.Extensions.DataIngestion.Tests/Writers/VectorStoreWriterTests.cs @@ -117,5 +117,56 @@ public async Task DoesSupportIncrementalIngestion() Assert.Equal("value2", record["key1"]); } + [Fact] + public async Task IncrementalIngestion_WithManyRecords_DeletesAllPreExistingChunks() + { + string documentId = Guid.NewGuid().ToString(); + + using TestEmbeddingGenerator testEmbeddingGenerator = new(); + using VectorStore vectorStore = CreateVectorStore(testEmbeddingGenerator); + using VectorStoreWriter writer = new( + vectorStore, + dimensionCount: TestEmbeddingGenerator.DimensionCount, + options: new() + { + IncrementalIngestion = true, + }); + + IngestionDocument document = new(documentId); + + // Create more chunks than the MaxTopCount (1000) to test pagination + // We create 2500 chunks to ensure multiple batches + List> chunks = []; + for (int i = 0; i < 2500; i++) + { + chunks.Add(new($"chunk {i}", document)); + } + + await writer.WriteAsync(chunks.ToAsyncEnumerable()); + + int recordCount = await writer.VectorStoreCollection + .GetAsync(filter: record => (string)record["documentid"]! == documentId, top: 10000) + .CountAsync(); + Assert.Equal(chunks.Count, recordCount); + + // Now we will do an incremental ingestion that should delete all pre-existing chunks + List> updatedChunks = + [ + new("updated chunk 1", document), + new("updated chunk 2", document) + ]; + + await writer.WriteAsync(updatedChunks.ToAsyncEnumerable()); + + // Verify that all old records were deleted and only the new ones remain + List> records = await writer.VectorStoreCollection + .GetAsync(filter: record => (string)record["documentid"]! == documentId, top: 10000) + .ToListAsync(); + + Assert.Equal(updatedChunks.Count, records.Count); + Assert.Contains(records, r => (string)r["content"]! == "updated chunk 1"); + Assert.Contains(records, r => (string)r["content"]! == "updated chunk 2"); + } + protected abstract VectorStore CreateVectorStore(TestEmbeddingGenerator testEmbeddingGenerator); }