[pangwangshu]

Addresses #40837.

Draft prototype of VespaWriteConfig for VectorDatabaseWriteTransform, following the design doc under discussion on dev@. It is shared so @LibenHailu and others can build on it; it is not ready for merge review until the dev@ questions are settled.

What it does

• Converts each EmbeddableItem to a Vespa document: id → id:<namespace>:<schema>::<id>, dense embedding → indexed tensor short form, sparse embedding → mapped tensor {"<index>": value}, content and metadata → fields (metadata_fields to rename/filter, fields_fn to override everything).
• Writes through Vespa's /document/v1 HTTP API with requests (already a core dependency), so there is no new dependency and no apache-beam[vespa] extra.
  • put → POST; update → PUT ?create=true with assign operations only, so bundle replays are idempotent.
  • Up to max_concurrent_requests documents in parallel over a pooled session.
  • 429 is retried with backoff and does not count against max_retries; 5xx, connection errors and timeouts are retried up to max_retries; other errors (e.g. a 400 for a field missing from the schema) fail the bundle at once with the document id and Vespa's message.
• Auth: mTLS cert/key file paths (checked on the worker in setup()) or a Vespa Cloud auth_token sent as a bearer token.

Testing

• vespa_test.py: 45 unit tests (config validation, converter, HTTP behavior and retries via requests_mock, pipeline with TestPipeline).
• vespa_it_test.py: 8 integration tests against vespaengine/vespa in testcontainers. They deploy a zipped application package through the config server and read documents back over /document/v1 (about 45s locally). The design doc plans this for a separate PR; it is included here so collaborators can run it.

The CHANGES.md entry and docs are deferred to a later PR, per the doc's rollout plan.

Open questions

From the design doc (pending on dev@):

1. Thin /document/v1 client (this PR) vs. pyvespa as an optional extra.
2. Which suite the container-based IT runs in.
3. Fail the bundle (this PR) vs. a failed-documents output for a dead-letter sink.

Found while implementing, open for discussion:
4. 429 retries are unbounded per the doc. Should there be a total time limit per document so a constantly overloaded cluster can't hang a batch job?
5. Throughput: one HTTP/1.1 request per document. Worth benchmarking against pyvespa's feed client before settling question 1.
6. One requests.Session is shared across the worker threads. This is common practice but not documented as thread-safe; one session per thread is an alternative.
7. Should the Vespa server-side timeout query parameter be sent alongside timeout_s?
8. update keeps fields it does not write (e.g. with content_field=None or an item without a sparse embedding), while put clears them. Document this, or leave it as expected partial-update behavior?
9. Backoff constants (0.5s initial, 30s cap) are hard-coded. Should they be configurable?
10. auth_token is serialized with the pipeline, like Qdrant's api_key. Is there a preferred way to pass secrets to IO configs?

🤖 Generated with Claude Code