xpark.dataset.connectors.InsertVectorDB#

class xpark.dataset.connectors.InsertVectorDB(connection_factory: Callable[[], Any], collection_name: str, metadata_columns: list[str | None] | None = None, **kwargs: Any)[source]#

Operator for batch-inserting records into a vector database.

Supported backends: [‘elasticsearch’, ‘milvus’].

IDs are auto-generated (UUID4) by this operator — callers should not supply an ID column. Internally uses the backend’s native upsert API (since the auto-generated UUID4 IDs guarantee no conflicts, upsert is equivalent to insert but is idempotent under retries). If you need to control record IDs, use UpsertVectorDB instead.

The operator is a pure side-effect: it returns a boolean column and does not produce meaningful output data.

Parameters:

  • connection_factory – A callable that returns a vector database client (e.g. lambda: pymilvus.MilvusClient(uri="...")). The backend is auto-detected from the returned client type.

  • collection_name – Name of the target collection.

  • metadata_columns (list[str | None] | None) – Maps each metadata column passed via .with_column() to a field name, like SQL AS. Each entry can be a str (use as the metadata field name) or None (parse the column value as JSON and merge all its key-value pairs into the metadata dict). Keys are case-sensitive and must be unique. Defaults to None.

  • **connector_kwargs – Extra keyword arguments. The connector-init reserved keys max_retries, retry_delay, retry_backoff are consumed by the connector itself; all remaining kwargs are forwarded to the underlying upsert call (e.g. timeout).

Examples

import pymilvus
from xpark.dataset import from_items
from xpark.dataset.connectors import InsertVectorDB
from xpark.dataset.expressions import col

ds = from_items([
    {"vec": [0.1, 0.2, 0.3], "category": "tech",
     "extra": '{"color": "red", "priority": 1}'},
])
ds = ds.with_column(
    "result",
    InsertVectorDB(
        connection_factory=lambda: pymilvus.MilvusClient(uri="http://localhost:19530"),
        collection_name="my_col",
        metadata_columns=["category", None],
    )
    .options(num_workers={"IO": 1})
    .with_column(col("vec"), col("category"), col("extra")),
)
# metadata => {"category": "tech", "color": "red", "priority": 1}
print(ds.take_all())

Methods

__call__(vectors, *metadata_cols)

Execute batch insert.

options(**kwargs)

with_column(vectors, *metadata_cols)

Execute batch insert.
__call__(vectors: pa.ChunkedArray, *metadata_cols: pa.ChunkedArray) → pa.Array#

Execute batch insert.

Parameters:

  • vectors – A ChunkedArray of list vectors.

  • *metadata_cols – Optional metadata columns.

Returns:

A boolean array (one False per row). The operator is used for its side-effects only.

options(**kwargs: Unpack[ExprUDFOptions]) → Self#
with_column(vectors: pa.ChunkedArray, *metadata_cols: pa.ChunkedArray) → pa.Array#

Execute batch insert.

Parameters:

  • vectors – A ChunkedArray of list vectors.

  • *metadata_cols – Optional metadata columns.

Returns:

A boolean array (one False per row). The operator is used for its side-effects only.