Skip to content
Merged
Show file tree
Hide file tree
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -169,7 +169,13 @@ private async Task<IReadOnlyList<object>> GetPreExistingChunksIdsAsync(Ingestion
}

// Each Vector Store has a different max top count limit, so we use low value and loop.
const int MaxTopCount = 1_000;
// Use smaller batch size in debug to be able to test the looping logic without needing to insert a lot of records.
const int MaxTopCount =
#if RELEASE
Comment thread
adamsitnik marked this conversation as resolved.
Outdated
1_000;
#else
10;
#endif

List<object> keys = [];
int insertedCount;
Expand All @@ -180,6 +186,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]!);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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<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 in debug builds (10) to test pagination
// In debug builds, MaxTopCount is 10, so we create 25 chunks to ensure multiple batches
List<IngestionChunk<string>> chunks = [];
for (int i = 0; i < 25; 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: 1000)
.CountAsync();
Assert.Equal(chunks.Count, recordCount);

// Now we will do an incremental ingestion that should delete all 25 pre-existing chunks
List<IngestionChunk<string>> 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<Dictionary<string, object?>> records = await writer.VectorStoreCollection
.GetAsync(filter: record => (string)record["documentid"]! == documentId, top: 1000)
.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);
}
Loading