Backport GetPreExistingChunksIdsAsync pagination fix

src/Libraries/Microsoft.Extensions.DataIngestion/Writers/VectorStoreWriter.cs

@@ -202,6 +202,7 @@ private async Task<IReadOnlyList<object>> 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]!);

test/Libraries/Microsoft.Extensions.DataIngestion.Tests/Writers/VectorStoreWriterTests.cs

@@ -152,5 +152,56 @@ public async Task BatchesChunks(int? batchTokenCount, int[] chunkTokenCounts)
         Assert.Equal(chunks.Count, recordCount);
     }
 
+    [Fact]
+    public async Task IncrementalIngestion_WithManyRecords_DeletesAllPreExistingChunks()
+    {
+        string documentId = Guid.NewGuid().ToString();
+
+        using TestEmbeddingGenerator<string> testEmbeddingGenerator = new();
+        using VectorStore vectorStore = CreateVectorStore(testEmbeddingGenerator);
+        using VectorStoreWriter<string> writer = new(
+            vectorStore,
+            dimensionCount: TestEmbeddingGenerator<string>.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<IngestionChunk<string>> chunks = [];
+        for (int i = 0; i < 2500; i++)
+        {
+            chunks.Add(TestChunkFactory.CreateChunk($"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<IngestionChunk<string>> updatedChunks =
+        [
+            TestChunkFactory.CreateChunk("updated chunk 1", document),
+            TestChunkFactory.CreateChunk("updated chunk 2", document)
+        ];
+
+        await writer.WriteAsync(updatedChunks.ToAsyncEnumerable());
+
+        // Verify that all old records were deleted and only the new ones remain
+        List<Dictionary<string, object?>> 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<string> testEmbeddingGenerator);
 }