[Python] Add Qdrant search enrichment handler

[pangwangshu]

Adds QdrantSearchEnrichmentHandler to apache_beam/ml/rag/enrichment, so a pipeline that ingests into Qdrant with QdrantWriteConfig can use Enrichment for the retrieval step. Until now the enrichment handlers covered only BigQuery and Milvus.

This builds on #35686 by @anush008, which was closed as stale before review. Anush agreed to my picking it up (comment) and is credited as a co-author on the commit.

Design

The handler follows MilvusSearchEnrichmentHandler, so the two read the same way:

Milvus	Qdrant
VectorSearchParameters	QdrantDenseSearchParameters
KeywordSearchParameters	QdrantSparseSearchParameters
HybridSearchParameters + ranker	QdrantHybridSearchParameters + fusion (RRF default, or DBSF)
MilvusSearchParameters	QdrantSearchParameters
MilvusSearchResult (id, distance, fields)	QdrantSearchResult (id, score, payload)

• Hybrid search sends the dense and sparse queries as prefetches and fuses the results server side through the Qdrant Query API.
• Each strategy accepts a payload filter, limit, score_threshold, search_params, and kwargs for other QueryRequest fields. with_payload can be a bool or a list of payload keys.
• QdrantHybridSearchParameters also takes a common filter, which Qdrant applies to both prefetches before their limits, and a score_threshold on the fused score.
• QdrantConnectionParameters is reused from ingestion/qdrant.py and re-exported. The default vector names (dense / sparse) match QdrantWriteConfig, so a collection written by Beam can be searched without extra configuration.
• Each batch from BatchElements is sent as a single query_batch_points request rather than one request per element.
• The __call__ return shape, empty-input handling and batch_elements_kwargs match the Milvus handler. join_fn also stores results under metadata["enrichment_data"], but returns a copy of the item instead of mutating the input element.
• Quota and timeout errors from REST (429, 408, 504, HTTP timeouts) and gRPC (RESOURCE_EXHAUSTED, DEADLINE_EXCEEDED) are raised as UserCodeQuotaException / UserCodeTimeoutException, so the Enrichment repeater retries them. Other errors propagate unchanged.
• Sparse search requires a sparse embedding. Unlike Milvus, Qdrant has no built-in BM25 over raw text.

Compared with #35686: the handler uses EmbeddableItem instead of the old Chunk type, reuses the existing connection parameters instead of duplicating them, and batches queries. No setup.py change is needed, because qdrant-client is already part of the qdrant extra.

A website example page, like the one for Milvus enrichment, can follow in a separate PR.

Testing

• qdrant_search_test.py: 40 unit tests covering parameter validation, plus dense, sparse, hybrid (RRF and DBSF, common filter, fused score threshold), filter, score_threshold, with_payload, batching, empty input, missing-embedding and unsupported-strategy errors, error mapping with the default repeater, and join_fn not mutating its input. They run against Qdrant's in-memory mode, so they need no Docker. All pass locally.
• qdrant_search_it_test.py: 7 integration tests. Each writes a corpus with QdrantWriteConfig(content_payload_key="content") to a testcontainers Qdrant instance, then reads it back through Enrichment (dense, filtered dense, sparse, hybrid, hybrid with a common filter and fused score threshold, batched, empty input). All pass locally on DirectRunner.
• yapf, pylint and ruff pass on the new files.
• CHANGES.md updated.

AI assistance disclosure

Per the ASF generative tooling guidance: this change was written with the help of Claude Code (Claude Opus 5.5). I have reviewed the code and tests and ran them locally, and to my knowledge the change contains no third-party copyrighted material.

Generated-by: Claude Code (Claude Opus 5.5)

Fixes #40628

🤖 Generated with Claude Code

[github-actions]

📋 Review Summary

This PR adds QdrantSearchEnrichmentHandler and associated configuration dataclasses (QdrantDenseSearchParameters, QdrantSparseSearchParameters, QdrantHybridSearchParameters, QdrantSearchParameters) to support dense, sparse, and server-side hybrid vector search enrichment against Qdrant collections. The implementation cleanly aligns with the existing MilvusSearchEnrichmentHandler and QdrantWriteConfig patterns and includes comprehensive unit and integration tests.

🔍 General Feedback

• The design, API surface, default vector naming, and batching behavior via query_batch_points are well-structured and idiomatic.
• Two minor edge cases in hybrid search filtering and mutable input mutation in join_fn are worth addressing.

---

💡 Tip: Comment /gemini review on this pull request to trigger a new Gemini code review.

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 79.97%. Comparing base (76f6e6d) to head (d84f8e5).
⚠️ Report is 211 commits behind head on master.

Additional details and impacted files

@@              Coverage Diff              @@
##             master   #40629       +/-   ##
=============================================
+ Coverage     56.16%   79.97%   +23.81%     
=============================================
  Files          1124      489      -635     
  Lines        178336    82730    -95606     
  Branches       1489        0     -1489     
=============================================
- Hits         100170    66167    -34003     
+ Misses        75645    16563    -59082     
+ Partials       2521        0     -2521     

Flag	Coverage Δ
go	?
java	?
python	79.97% <100.00%> (+0.42%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[github-actions]

Assigning reviewers:

R: @shunping for label python.
Backup reviewer: claudevdm (if the main reviewer is unresponsive, you can escalate to them by tagging their username)

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).