diff --git a/python/python/lancedb/query.py b/python/python/lancedb/query.py index 80927093c..cef89e63c 100644 --- a/python/python/lancedb/query.py +++ b/python/python/lancedb/query.py @@ -78,6 +78,10 @@ if TYPE_CHECKING: T = TypeVar("T", bound="LanceModel") AnalyzePlanDistributedMetrics = Literal["aggregate", "per_worker", "full"] +# Number of rows a hybrid query returns when no limit was set on it. This +# mirrors the default the Rust query builder applies to its sub-queries. +DEFAULT_HYBRID_LIMIT = 10 + @runtime_checkable class _LanceScanner(Protocol): @@ -3893,14 +3897,54 @@ class AsyncHybridQuery(AsyncStandardQuery, AsyncVectorQueryBase): return self + def _create_child_queries( + self, + ) -> Tuple["AsyncFTSQuery", "AsyncVectorQuery", int, int]: + """Build the sub-queries that make up this hybrid query. + + Execution, `explain_plan` and `analyze_plan` all go through here so that + the plans that are reported are the plans that actually run. + + Returns the two sub-queries along with the effective limit and offset of + the hybrid query itself. + """ + fts_query = AsyncFTSQuery(self._inner.to_fts_query(), self._table) + vec_query = AsyncVectorQuery(self._inner.to_vector_query(), self._table) + + fts_req = fts_query._inner.to_query_request() + vec_req = vec_query._inner.to_query_request() + + # Only one of the two sub-queries carries the limit when it was never + # set explicitly: nearest_to()/nearest_to_text() build the sibling query + # from scratch, and that is where the default gets filled in. Which one + # that is depends on the order the hybrid query was built in, so look at + # both rather than at a single side. + limit = fts_req.limit if fts_req.limit is not None else vec_req.limit + if limit is None: + limit = DEFAULT_HYBRID_LIMIT + offset = fts_req.offset or vec_req.offset or 0 + + fts_query.with_row_id() + vec_query.with_row_id() + + # offset() pushes the offset down into both sub-queries, which would make + # each of them skip its own first `offset` rows. The window has to be + # taken out of the combined, reranked results instead, so fetch the + # skipped prefix here too and slice it off afterwards. + fts_query.limit(limit + offset) + vec_query.limit(limit + offset) + fts_query.offset(0) + vec_query.offset(0) + + return fts_query, vec_query, limit, offset + async def to_batches( self, *, max_batch_length: Optional[int] = None, timeout: Optional[timedelta] = None, ) -> AsyncRecordBatchReader: - fts_query = AsyncFTSQuery(self._inner.to_fts_query(), self._table) - vec_query = AsyncVectorQuery(self._inner.to_vector_query(), self._table) + fts_query, vec_query, limit, offset = self._create_child_queries() req = fts_query._inner.to_query_request() blob_auto_row_id = False @@ -3920,9 +3964,6 @@ class AsyncHybridQuery(AsyncStandardQuery, AsyncVectorQueryBase): self._blob_auto_row_id = blob_auto_row_id self._blob_paths = blob_paths - fts_query.with_row_id() - vec_query.with_row_id() - fts_results, vector_results = await asyncio.gather( fts_query.to_arrow(timeout=timeout), vec_query.to_arrow(timeout=timeout), @@ -3934,8 +3975,9 @@ class AsyncHybridQuery(AsyncStandardQuery, AsyncVectorQueryBase): norm=self._norm, fts_query=fts_query.get_query(), reranker=self._reranker, - limit=self._inner.get_limit(), + limit=limit, with_row_ids=True, + offset=offset, ) if ( not self._user_requested_row_id() @@ -3964,14 +4006,14 @@ class AsyncHybridQuery(AsyncStandardQuery, AsyncVectorQueryBase): ... print(plan) >>> asyncio.run(doctest_example()) # doctest: +ELLIPSIS, +NORMALIZE_WHITESPACE RRFReranker(K=60) - ProjectionExec: expr=[vector@0 as vector, text@3 as text, _distance@2 as _distance] + ProjectionExec: expr=[vector@0 as vector, text@3 as text, _distance@2 as _distance, _rowid@1 as _rowid] LanceRead: uri=..., projection=[text], source=stream(_rowid) GlobalLimitExec: skip=0, fetch=10 FilterExec: _distance@2 IS NOT NULL SortExec: TopK(fetch=10), expr=[_distance@2 ASC NULLS LAST, _rowid@1 ASC NULLS LAST], preserve_partitioning=[false] KNNVectorDistance: metric=l2 LanceRead: uri=..., projection=[vector], ... - ProjectionExec: expr=[vector@2 as vector, text@3 as text, _score@1 as _score] + ProjectionExec: expr=[vector@2 as vector, text@3 as text, _score@1 as _score, _rowid@0 as _rowid] LanceRead: uri=..., projection=[vector, text], source=stream(_rowid) GlobalLimitExec: skip=0, fetch=10 MatchQuery: column=text, query=[hello] @@ -3986,8 +4028,9 @@ class AsyncHybridQuery(AsyncStandardQuery, AsyncVectorQueryBase): plan : str """ # noqa: E501 - vector_plan = await self._inner.to_vector_query().explain_plan(verbose) - fts_plan = await self._inner.to_fts_query().explain_plan(verbose) + fts_query, vec_query, _, _ = self._create_child_queries() + vector_plan = await vec_query.explain_plan(verbose) + fts_plan = await fts_query.explain_plan(verbose) # Indent sub-plans under the reranker indented_vector = "\n".join(" " + line for line in vector_plan.splitlines()) indented_fts = "\n".join(" " + line for line in fts_plan.splitlines()) @@ -4014,14 +4057,12 @@ class AsyncHybridQuery(AsyncStandardQuery, AsyncVectorQueryBase): ------- plan : str """ + fts_query, vec_query, _, _ = self._create_child_queries() + results = ["Vector Search Query:"] - results.append( - await self._inner.to_vector_query().analyze_plan(distributed_metrics) - ) + results.append(await vec_query.analyze_plan(distributed_metrics)) results.append("FTS Search Query:") - results.append( - await self._inner.to_fts_query().analyze_plan(distributed_metrics) - ) + results.append(await fts_query.analyze_plan(distributed_metrics)) return "\n".join(results) diff --git a/python/python/tests/test_hybrid_query.py b/python/python/tests/test_hybrid_query.py index 5e9b45ecb..61b8001cf 100644 --- a/python/python/tests/test_hybrid_query.py +++ b/python/python/tests/test_hybrid_query.py @@ -203,6 +203,93 @@ async def test_async_hybrid_query_default_limit(table: AsyncTable): assert texts.count("a") == 1 +@pytest.mark.asyncio +async def test_async_hybrid_query_offset(table: AsyncTable): + # The offset window of a hybrid query must be a suffix of the same query + # run without an offset. Skipping the first rows of each sub-query instead + # of the first rows of the fused result silently changes which rows land in + # the window. + full = await ( + table.query() + .nearest_to([0.0, 0.4]) + .nearest_to_text("dog") + .limit(4) + .with_row_id() + .to_arrow() + ) + assert len(full) == 4 + + second_page = await ( + table.query() + .nearest_to([0.0, 0.4]) + .nearest_to_text("dog") + .offset(2) + .limit(2) + .with_row_id() + .to_arrow() + ) + assert second_page["_rowid"].to_pylist() == full["_rowid"].to_pylist()[2:] + + first_page = await ( + table.query() + .nearest_to([0.0, 0.4]) + .nearest_to_text("dog") + .limit(2) + .with_row_id() + .to_arrow() + ) + # Paging through the result must visit every row exactly once: no row + # repeated from the previous page and none dropped between the two. + paged = first_page["_rowid"].to_pylist() + second_page["_rowid"].to_pylist() + assert sorted(paged) == sorted(full["_rowid"].to_pylist()) + + +@pytest.mark.asyncio +async def test_async_hybrid_query_fts_first_default_limit(table: AsyncTable): + # nearest_to() and nearest_to_text() build their new sibling sub-query from + # scratch, and that is the sub-query the default limit ends up on. So the + # side that carries the limit depends on the order the hybrid query was + # built in, and looking at only one side loses the limit for half the ways + # a hybrid query can be written. Without a limit the combined results are + # not truncated at all and the whole union of both candidate lists is + # returned. + await table.add([{"text": "dog", "vector": [50.0 + i, 50.0]} for i in range(10)]) + + result = await ( + table.query().nearest_to_text("dog").nearest_to([0.1, 0.1]).to_arrow() + ) + assert len(result) == 10 + + offset_result = await ( + table.query().nearest_to_text("dog").nearest_to([0.1, 0.1]).offset(2).to_arrow() + ) + assert len(offset_result) == 10 + + +@pytest.mark.asyncio +async def test_async_hybrid_query_explain_plan_matches_execution(table: AsyncTable): + # Paging rewrites the sub-queries: each one fetches limit + offset rows with + # no offset of its own, and the window is sliced out after fusion. The plans + # have to be built from those rewritten sub-queries, otherwise explain_plan + # and analyze_plan describe a query that is never run. + query = ( + table.query().nearest_to([0.0, 0.4]).nearest_to_text("dog").offset(2).limit(2) + ) + await query.to_arrow() + + plan = await query.explain_plan() + assert [ + line.strip() for line in plan.splitlines() if "GlobalLimitExec" in line + ] == [ + "GlobalLimitExec: skip=0, fetch=4", + "GlobalLimitExec: skip=0, fetch=4", + ] + + analyzed = await query.analyze_plan() + assert analyzed.count("skip=0, fetch=4") == 2 + assert "skip=2" not in analyzed + + def test_hybrid_query_offset(sync_table: Table): # The offset window of a hybrid query must be a suffix of the same query # run without an offset -- it must not be silently ignored.