WeaviateEx.Batch.VectorIndexing (WeaviateEx v0.7.4)

View Source

Wait for vector indexing completion after batch operations.

When objects are inserted via batch operations, their vectors are indexed asynchronously. This module provides functions to wait for indexing completion, ensuring all vectors are queryable before continuing.

Features

  • Poll shard status until vectors are indexed
  • Configurable timeout and poll interval
  • Support for specific shards or all collection shards
  • Failure tolerance with configurable retry limits

Example

alias WeaviateEx.Batch.VectorIndexing

# After batch insert, wait for indexing
{:ok, _} = WeaviateEx.Batch.create_objects(client, objects)
:ok = VectorIndexing.wait_for_indexing(client, "Article")

# Wait for specific tenant shard
:ok = VectorIndexing.wait_for_indexing(client, "Article",
  tenant: "tenant-a",
  timeout: 120_000,
  poll_interval: 500
)

# Check if a specific shard is ready
{:ok, true} = VectorIndexing.shard_ready?(client, "Article", "shard-0")

Summary

Types

opts()
shard_spec()

Functions

all_indexed?(client, collection)

Check if all vectors are indexed for a collection.

get_queue_status(client, collection)

Get the vector queue status for all shards in a collection.

shard_ready?(client, collection, shard_name)

Check if a shard is ready for queries.

total_pending_vectors(client, collection)

Get the total pending vector count across all shards.

wait_for_indexing(client, collection, opts \\ [])

Wait for all vectors to be indexed.

wait_for_shards(client, shards, opts \\ [])

Wait for specific shards to have vectors indexed.

Types

opts()

@type opts() :: keyword()

shard_spec()

@type shard_spec() :: %{collection: String.t(), tenant: String.t() | nil}

Functions

all_indexed?(client, collection)

@spec all_indexed?(WeaviateEx.Client.t(), String.t()) ::
  {:ok, boolean()} | {:error, WeaviateEx.Error.t()}

Check if all vectors are indexed for a collection.

Examples

{:ok, true} = VectorIndexing.all_indexed?(client, "Article")

Returns

  • {:ok, boolean()} - Whether all vectors are indexed
  • {:error, Error.t()} - Error fetching status

get_queue_status(client, collection)

@spec get_queue_status(WeaviateEx.Client.t(), String.t()) ::
  {:ok, [WeaviateEx.Cluster.Shard.t()]} | {:error, WeaviateEx.Error.t()}

Get the vector queue status for all shards in a collection.

Returns a list of shards with their current vector queue sizes.

Examples

{:ok, statuses} = VectorIndexing.get_queue_status(client, "Article")
for status <- statuses do
  IO.puts("#{status.name}: #{status.vector_queue_size} vectors pending")
end

Returns

  • {:ok, [Shard.t()]} - List of shards with queue status
  • {:error, Error.t()} - Error fetching status

shard_ready?(client, collection, shard_name)

@spec shard_ready?(WeaviateEx.Client.t(), String.t(), String.t()) ::
  {:ok, boolean()} | {:error, WeaviateEx.Error.t()}

Check if a shard is ready for queries.

A shard is ready when its vector queue is empty, meaning all vectors have been indexed.

Examples

{:ok, true} = VectorIndexing.shard_ready?(client, "Article", "shard-0")
{:ok, false} = VectorIndexing.shard_ready?(client, "Article", "shard-0")

Returns

  • {:ok, boolean()} - Whether shard is ready
  • {:error, Error.t()} - Error fetching shard status

total_pending_vectors(client, collection)

@spec total_pending_vectors(WeaviateEx.Client.t(), String.t()) ::
  {:ok, non_neg_integer()} | {:error, WeaviateEx.Error.t()}

Get the total pending vector count across all shards.

Examples

{:ok, 1500} = VectorIndexing.total_pending_vectors(client, "Article")

Returns

  • {:ok, non_neg_integer()} - Total pending vectors
  • {:error, Error.t()} - Error fetching status

wait_for_indexing(client, collection, opts \\ [])

@spec wait_for_indexing(WeaviateEx.Client.t(), String.t(), opts()) ::
  :ok | {:error, term()}

Wait for all vectors to be indexed.

Polls shard status until all shards have an empty vector queue, indicating all vectors have been indexed and are queryable.

Options

  • :timeout - Maximum wait time in milliseconds (default: 60_000)
  • :poll_interval - Milliseconds between status checks (default: 250)
  • :how_many_failures - Number of consecutive failures before giving up (default: 5)
  • :tenant - Filter to specific tenant (optional)

Examples

# Wait for all shards in a collection
:ok = VectorIndexing.wait_for_indexing(client, "Article")

# Wait with custom timeout
:ok = VectorIndexing.wait_for_indexing(client, "Article", timeout: 120_000)

# Wait for specific tenant
:ok = VectorIndexing.wait_for_indexing(client, "Article", tenant: "tenant-a")

Returns

  • :ok - All vectors indexed
  • {:error, :timeout} - Timed out waiting for indexing
  • {:error, {:max_failures_exceeded, last_error}} - Too many consecutive failures

wait_for_shards(client, shards, opts \\ [])

@spec wait_for_shards(WeaviateEx.Client.t(), [shard_spec()], opts()) ::
  :ok | {:error, term()}

Wait for specific shards to have vectors indexed.

Similar to wait_for_indexing/3 but accepts a list of shard specifications.

Examples

shards = [
  %{collection: "Article", tenant: nil},
  %{collection: "Product", tenant: "tenant-a"}
]
:ok = VectorIndexing.wait_for_shards(client, shards)

Returns

  • :ok - All specified shards have vectors indexed
  • {:error, :timeout} - Timed out waiting
  • {:error, term()} - Other error