[Feature Request]: Vespa ingestion (VespaWriteConfig) for Python ml.rag #40837

What would you like to happen?

Add a VespaWriteConfig in apache_beam.ml.rag.ingestion.vespa that plugs into the existing VectorDatabaseWriteTransform, so a Beam RAG pipeline can write EmbeddableItems (content, metadata, dense and sparse embeddings) into a Vespa schema.

ml.rag ingestion already covers BigQuery, AlloyDB / Cloud SQL / Postgres, MySQL, Spanner, Milvus and Qdrant. Vespa is widely used for hybrid search (BM25, nearest-neighbor and custom ranking in one query), and right now its users have to write their own DoFn. Vespa accepts dense and sparse tensors natively, so both fields of Embedding map over without loss.

Design doc: https://docs.google.com/document/d/1cC5wezcXugmyHDbt-xuP6zrlKia337dJObWNKnZZJvM/edit
dev@ discussion: https://lists.apache.org/thread/7jr1l0ojljzdt3p13396nzp2vlbzzlkx

Proposal summary

• API: VespaConnectionParameters (url, port, mTLS cert/key, Vespa Cloud token, extra headers) and VespaWriteConfig(VectorDatabaseWriteConfig) with schema, namespace, content_field, dense_embedding_field, sparse_embedding_field, metadata_fields, an optional fields_fn override, operation_type, and feed tuning knobs (queue size, workers, connections, 429 retries).
• Field mapping:
  • EmbeddableItem.id becomes the user part of id:<namespace>:<schema>::<id>.
  • Dense embeddings are written as a JSON array (indexed tensor short form).
  • Sparse embeddings are written as a {"<index>": value} map (mapped tensor short form).
  • Content and metadata go to top-level fields. A field name collision raises ValueError, the same as the Qdrant config.
• Write path: a Map converts each item, then a DoFn buffers per bundle and flushes with pyvespa's feed_iterable. Failed documents make the bundle fail so the runner retries it. Puts are idempotent by document id, so a retry doesn't create duplicates. The client is created in setup() and closed in teardown(), as in the Qdrant writer.
• Dependency: pyvespa (Apache-2.0) as a new optional extra apache-beam[vespa], imported lazily.
• Testing: unit tests with a mocked client, plus an integration test that starts vespaengine/vespa with a generic testcontainers DockerContainer, deploys a minimal application package, feeds documents and reads them back by id.
• Out of scope: deploying or changing Vespa application packages. A Vespa enrichment handler for retrieval is planned as a follow-up PR.

Open questions

1. Is Vespa a welcome addition to ml.rag ingestion?
2. Should we depend on pyvespa as an optional extra (it pulls in docker, lxml, fastcore, httpx), or write a thin client over Vespa's /document/v1 HTTP API with requests?
3. Where should the container-based integration test run: in the same ML suites as the Milvus/Qdrant ITs, or postcommit only? (Container start plus deploy takes tens of seconds.)
4. What should the default operation be: put (replace the whole document) or partial update with create=true?

I'm happy to implement this. Feedback on the questions above is welcome here or on the dev@ thread.

Issue Priority

Priority: 2 (default / most feature requests should be filed as P2)

Issue Components

• Component: Python SDK
• Component: Java SDK
• Component: Go SDK
• Component: Typescript SDK
• Component: IO connector
• Component: Beam YAML
• Component: Beam examples
• Component: Beam playground
• Component: Beam katas
• Component: Website
• Component: Infrastructure
• Component: Spark Runner
• Component: Flink Runner
• Component: Prism Runner
• Component: Twister2 Runner
• Component: Hazelcast Jet Runner
• Component: Google Cloud Dataflow Runner