[Feature Request]: Optionally write chunk content to the payload in QdrantWriteConfig #40476

What would you like to happen?

QdrantWriteConfig (added in #38142) currently stores only item.metadata as the point payload; item.content (the chunk text) is not written anywhere:

https://github.com/apache/beam/blob/master/sdks/python/apache_beam/ml/rag/ingestion/qdrant.py

return models.PointStruct(
    id=id,
    vector=vector,
    payload=item.metadata if item.metadata else None,
)

The other RAG writers do persist the content: BigQueryVectorWriterConfig writes a content column (item.content_string) and the Milvus writer has with_content_spec(). With Qdrant, a search over a collection written by Beam returns vectors and metadata but not the text, so the retrieved chunks cannot be used for generation without a second lookup against the original source.

Proposal

Add an optional field to QdrantWriteConfig:

content_payload_key: Optional[str] = None

• Default None keeps today's behavior exactly (no change for existing users).
• When set, the converter stores item.content_string under that key in the payload, alongside the metadata. A collision with an existing metadata key should raise a clear ValueError rather than silently overwrite.
• Unit tests for the converter plus an update to the existing testcontainers round-trip test.

Questions for maintainers

1. Was omitting the content intentional (e.g. to keep payloads small)? The existing tests assert payload == item.metadata, so I want to make sure this is not a deliberate design decision before changing it.
2. Is an opt-in content_payload_key the right shape, or would you prefer a different name/default (e.g. always writing to a "content" key)?

Context

I plan to follow up with a Qdrant search enrichment handler (the earlier attempt in #35686 went stale), and the retrieved text is needed for that to be useful end to end. I would like to send this smaller change first, and I'm happy to take it on if the direction looks good.

Issue Priority

Priority: 3 (nice-to-have improvement)

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