Skip to content

Document server-side batching ingest and refactor the batch import page #478

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
g-despot wants to merge 3 commits into
base: main
Choose a base branch
from
Open

Document server-side batching ingest and refactor the batch import page #478

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
7d7d6a8
Document server-side batching ingest and refactor the batch import page
g-despot Jul 22, 2026
a925388
Address review: differentiate TypeScript SSB examples, drop extras
g-despot Jul 22, 2026
96e58df
Scope language-specific prose to the language tabs
g-despot Jul 22, 2026
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
80 changes: 67 additions & 13 deletions _includes/code/csharp/ManageObjectsImportTest.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@
using System.Threading.Tasks;
using CsvHelper;
using Weaviate.Client;
using Weaviate.Client.Batch;
using Weaviate.Client.Models;
using Xunit;

Expand Down Expand Up @@ -146,8 +147,55 @@ await client.Collections.Create(

var collection = client.Collections.Use("MyCollection");

// Use `Batch.InsertMany` for server-side batching. The client sends
// data in batches at a rate controlled by the server.
// Use `Batch.StartBatch` for server-side batching. The client streams
// objects to the server, which paces the import based on its own load.
// highlight-start
await using var batch = await collection.Batch.StartBatch();

var handles = new List<TaskHandle>();
foreach (var dataRow in dataRows)
{
handles.Add(await batch.Add(dataRow));
}

await batch.Close();
// highlight-end

var results = await Task.WhenAll(handles.Select(h => h.Result));
var failedObjects = results.Where(r => !r.Success).ToList();
if (failedObjects.Any())
{
Console.WriteLine($"Number of failed imports: {failedObjects.Count}");
}
// END ServerSideBatchImportExample

var result = await collection.Aggregate.OverAll(totalCount: true);
Assert.Equal(5, result.TotalCount);
}

[Fact]
public async Task TestServerSideIngest()
{
await BeforeEach();
await client.Collections.Create(
new CollectionCreateParams
{
Name = "MyCollection",
VectorConfig = Configure.Vector("default", v => v.SelfProvided()),
}
);

// START ServerSideIngestExample
var dataRows = Enumerable
.Range(0, 5)
.Select(i => new { title = $"Object {i + 1}" })
.ToList();

var collection = client.Collections.Use("MyCollection");

// `Batch.InsertMany` is the one-shot server-side ingest of an
// in-memory list. The client streams the list to the server
// using server-side batching under the hood.
// highlight-start
var response = await collection.Batch.InsertMany(dataRows);
// highlight-end
Expand All @@ -158,7 +206,7 @@ await client.Collections.Create(
Console.WriteLine($"Number of failed imports: {failedObjects.Count}");
Console.WriteLine($"First failed object: {failedObjects.First().Error}");
}
// END ServerSideBatchImportExample
// END ServerSideIngestExample

var result = await collection.Aggregate.OverAll(totalCount: true);
Assert.Equal(5, result.TotalCount);
Expand Down Expand Up @@ -194,8 +242,9 @@ await client.Collections.Create(

var collection = client.Collections.Use("MyCollection");

// `Batch.InsertMany` imports the list using server-side batching.
// highlight-start
var response = await collection.Data.InsertMany(dataToInsert);
var response = await collection.Batch.InsertMany(dataToInsert);
// highlight-end

var failedObjects = response.Where(r => r.Error != null).ToList();
Expand Down Expand Up @@ -245,13 +294,17 @@ await client.Collections.Create(

var collection = client.Collections.Use("MyCollection");

var response = await collection.Data.InsertMany(dataToInsert);
// `Batch.InsertMany` imports the list using server-side batching.
// highlight-start
var response = await collection.Batch.InsertMany(dataToInsert);
// highlight-end

// Handle errors
if (response.HasErrors)
var failedObjects = response.Where(r => r.Error != null).ToList();
if (failedObjects.Any())
{
Console.WriteLine($"Number of failed imports: {response.Errors.Count()}");
Console.WriteLine($"First failed object: {response.Errors.First().Message}");
Console.WriteLine($"Number of failed imports: {failedObjects.Count}");
Console.WriteLine($"First failed object: {failedObjects.First().Error}");
}
// END BatchImportWithVectorExample

Expand Down Expand Up @@ -342,16 +395,17 @@ await client.Collections.Create(

var collection = client.Collections.Use("MyCollection");

// Insert the data using InsertMany
// `Batch.InsertMany` imports the list using server-side batching.
// highlight-start
var response = await collection.Data.InsertMany(dataToInsert);
var response = await collection.Batch.InsertMany(dataToInsert);
// highlight-end

// Handle errors
if (response.HasErrors)
var failedObjects = response.Where(r => r.Error != null).ToList();
if (failedObjects.Any())
{
Console.WriteLine($"Number of failed imports: {response.Errors.Count()}");
Console.WriteLine($"First failed object error: {response.Errors.First().Message}");
Console.WriteLine($"Number of failed imports: {failedObjects.Count}");
Console.WriteLine($"First failed object error: {failedObjects.First().Error}");
}
// END BatchImportWithNamedVectors
}
Expand Down
186 changes: 121 additions & 65 deletions _includes/code/howto/manage-data.import.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -157,38 +157,43 @@
# ===== Insert many with custom ID =====
# =======================================

# Re-create the collection
client.collections.delete("MyCollection")
client.collections.create(
"MyCollection",
vector_config=Configure.Vectors.self_provided()
)

# START BatchImportWithIDExample
# highlight-start
from weaviate.util import generate_uuid5 # Generate a deterministic ID
from weaviate.classes.data import DataObject
# highlight-end

# START BatchImportWithIDExample
data_rows = [{"title": f"Object {i+1}"} for i in range(5)]

collection = client.collections.use("MyCollection")

# highlight-start
with collection.batch.fixed_size(batch_size=200) as batch:
for data_row in data_rows:
obj_uuid = generate_uuid5(data_row)
batch.add_object(
properties=data_row,
uuid=obj_uuid
)
data_objects = [
DataObject(
properties=data_row,
uuid=generate_uuid5(data_row)
)
for data_row in data_rows
]

result = collection.data.ingest(data_objects)
# highlight-end
if batch.number_errors > 10:
print("Batch import stopped due to excessive errors.")
break

failed_objects = collection.batch.failed_objects
if failed_objects:
print(f"Number of failed imports: {len(failed_objects)}")
print(f"First failed object: {failed_objects[0]}")
if result.errors:
print(f"Number of failed imports: {len(result.errors)}")
# END BatchImportWithIDExample

result = collection.aggregate.over_all(total_count=True)
assert result.total_count == 5
resp_obj = collection.query.fetch_object_by_id(obj_uuid)
# Tests
agg_result = collection.aggregate.over_all(total_count=True)
assert agg_result.total_count == 5
resp_obj = collection.query.fetch_object_by_id(generate_uuid5(data_rows[-1]))
assert resp_obj != None
# Clean up
client.collections.delete(collection.name)
Expand All @@ -197,32 +202,40 @@
# ===== Batch import with custom vector =====
# ===========================================

# Re-create the collection
client.collections.delete("MyCollection")
client.collections.create(
"MyCollection",
vector_config=Configure.Vectors.self_provided()
)

# START BatchImportWithVectorExample
from weaviate.classes.data import DataObject

data_rows = [{"title": f"Object {i+1}"} for i in range(5)]
vectors = [[0.1] * 1536 for i in range(5)]

collection = client.collections.use("MyCollection")

# highlight-start
with collection.batch.fixed_size(batch_size=200) as batch:
for i, data_row in enumerate(data_rows):
batch.add_object(
properties=data_row,
vector=vectors[i]
)
data_objects = [
DataObject(
properties=data_row,
vector=vectors[i]
)
for i, data_row in enumerate(data_rows)
]

result = collection.data.ingest(data_objects)
# highlight-end
if batch.number_errors > 10:
print("Batch import stopped due to excessive errors.")
break

failed_objects = collection.batch.failed_objects
if failed_objects:
print(f"Number of failed imports: {len(failed_objects)}")
print(f"First failed object: {failed_objects[0]}")
if result.errors:
print(f"Number of failed imports: {len(result.errors)}")
# END BatchImportWithVectorExample

result = collection.aggregate.over_all(total_count=True)
assert result.total_count == 5
# Tests
agg_result = collection.aggregate.over_all(total_count=True)
assert agg_result.total_count == 5
# Clean up
client.collections.delete(collection.name)

Expand Down Expand Up @@ -255,6 +268,8 @@
)

# START BatchImportWithNamedVectors
from weaviate.classes.data import DataObject

data_rows = [{
"title": f"Object {i+1}",
"body": f"Body {i+1}"
Expand All @@ -266,24 +281,22 @@
collection = client.collections.use("MyCollection")

# highlight-start
with collection.batch.fixed_size(batch_size=200) as batch:
for i, data_row in enumerate(data_rows):
batch.add_object(
properties=data_row,
vector={
"title": title_vectors[i],
"body": body_vectors[i],
}
)
data_objects = [
DataObject(
properties=data_row,
vector={
"title": title_vectors[i],
"body": body_vectors[i],
}
)
for i, data_row in enumerate(data_rows)
]

result = collection.data.ingest(data_objects)
# highlight-end
if batch.number_errors > 10:
print("Batch import stopped due to excessive errors.")
break

failed_objects = collection.batch.failed_objects
if failed_objects:
print(f"Number of failed imports: {len(failed_objects)}")
print(f"First failed object: {failed_objects[0]}")
if result.errors:
print(f"Number of failed imports: {len(result.errors)}")
# END BatchImportWithNamedVectors

response = collection.query.fetch_objects(include_vector=True)
Expand Down Expand Up @@ -318,36 +331,37 @@
]
)

from_uuid = authors.data.insert(
properties={"name": "Jane Austen"}
)

publications.data.insert(
{"title": "Ye Olde Times"}
)
target_uuid = publications.query.fetch_objects(limit=1).objects[0].uuid

# BatchImportWithRefExample
# START BatchImportWithRefExample
from weaviate.classes.data import DataObject

collection = client.collections.use("Author")

with collection.batch.fixed_size(batch_size=100) as batch:
batch.add_reference(
from_property="writesFor",
from_uuid=from_uuid,
to=target_uuid,
)
# highlight-start
data_objects = [
DataObject(
properties={"name": "Jane Austen"},
references={"writesFor": target_uuid},
),
]

result = collection.data.ingest(data_objects)
# highlight-end

failed_references = collection.batch.failed_references
if failed_references:
print(f"Number of failed imports: {len(failed_references)}")
print(f"First failed reference: {failed_references[0]}")
if result.errors:
print(f"Number of failed imports: {len(result.errors)}")
# END BatchImportWithRefExample

# Tests
from weaviate.classes.query import QueryReference

new_uuid = result.uuids[0]
response = collection.query.fetch_object_by_id(
from_uuid,
new_uuid,
return_references=QueryReference(link_on="writesFor", return_properties=["title"])
)

Expand Down Expand Up @@ -602,4 +616,46 @@ def add_object(obj) -> None:
client.collections.delete(collection.name)


# ==================================================
# ===== Server-side one-shot ingest =====
# ==================================================

# Re-create the collection
client.collections.delete("MyCollection")
client.collections.create(
"MyCollection",
vector_config=Configure.Vectors.self_provided()
)

# START ServerSideIngestExample
data_rows = [
{"title": f"Object {i+1}"} for i in range(5)
]

collection = client.collections.use("MyCollection")

# highlight-start
# `ingest` imports the whole list with server-side batching in a single call
result = collection.data.ingest(data_rows)
# highlight-end

# The return object is the same as for `insert_many`
if result.errors:
print(f"Number of failed imports: {len(result.errors)}")
# `errors` is a dict keyed by the index of the failed object
for index, error in result.errors.items():
print(f"Failed object at index {index}: {error.message}")
# END ServerSideIngestExample

# Tests
assert len(result.errors) == 0
assert len(result.uuids) == 5

agg_result = collection.aggregate.over_all(total_count=True)
assert agg_result.total_count == 5

# Clean up
client.collections.delete(collection.name)


client.close()
Loading
Loading