fix(python): apply offset when combining async hybrid results (#4028)

Fixes #4027

## Summary

`AsyncHybridQuery`
(`table.query().nearest_to(...).nearest_to_text(...)`) paginates
incorrectly when `.offset()` is used: the second page repeats rows from
the first page and silently drops others.

`offset()` on a hybrid query pushes the offset down into *both*
sub-queries (`HybridQuery::offset` in `python/src/query.rs` forwards to
`inner_vec` and `inner_fts`), so each sub-query independently skips its
own first `offset` rows before the results are fused.
`AsyncHybridQuery.to_batches` then called `_combine_hybrid_results(...,
limit=self._inner.get_limit())` without an `offset`, so the reranked
table was sliced starting at position 0 and the sub-query limits were
never raised to cover the skipped prefix.

On the 4-row fixture in `test_hybrid_query.py`, with `_rowid` ordering
`[3, 0, 2, 1]`:

| query | before | after |
| --- | --- | --- |
| `.limit(2)` | `[0, 3]` | `[0, 3]` |
| `.offset(2).limit(2)` | `[3, 1]` | `[2, 1]` |

Row `3` was returned on both pages and row `2` was never returned at
all.

This is the async counterpart of #3769 (`Fixes #3765`), which fixed the
same bug in the synchronous `LanceHybridQueryBuilder`. #3765 explicitly
deferred the async path; this PR closes that gap and reuses the `offset`
parameter that #3769 already added to `_combine_hybrid_results`. The
synchronous path is unaffected — it was fixed in #3769.

## Changes

`python/python/lancedb/query.py`, `AsyncHybridQuery.to_batches`:

- Each sub-query now fetches `limit + offset` rows and its own offset is
reset to 0, so the fused result contains the full prefix the window is
sliced out of.
- The combined, reranked table is sliced with `offset=` instead of
always starting at 0.

Both halves are needed: raising the sub-query limits without the final
slice still returns page 1, and slicing without raising the limits still
misses rows.

`nodejs` has no equivalent hybrid combine path, so there is no SDK
parity gap here.

## Test plan

- [x] New regression test `test_async_hybrid_query_offset` in
`python/python/tests/test_hybrid_query.py`, mirroring the sync
`test_hybrid_query_offset`. It asserts the offset window is a suffix of
the un-offset result *and* that page 1 + page 2 together cover every row
exactly once (a row-count-only assertion would pass even with
duplicates).
- [x] `pytest python/tests/test_hybrid_query.py` — 16 passed
- [x] `pytest python/tests/test_rerankers.py` — 9 passed, 11 skipped
- [x] `pytest python/tests/test_query.py` — 86 passed
- [x] `pytest --doctest-modules python/lancedb/query.py` — 13 passed
- [x] `ruff format --check` / `ruff check` — clean
---

## Scope, after review

@lancedb-gatekeeper raised three points. Two were mine and are fixed in
`04d07c2`; the third is deliberately left alone and I'd like a
maintainer's call on it.

**Fixed — effective limit was read from the FTS child only.**
`HybridQuery::get_limit()` (`python/src/query.rs:1159`) returns
`self.inner_fts.inner.current_request().limit`, so an FTS-first hybrid
with no explicit `.limit()` yielded `None`, skipped the widening branch
and passed `limit=None` to the combiner — returning the union of both
candidate lists instead of the documented default of 10. The limit is
now derived from both children with a `DEFAULT_HYBRID_LIMIT = 10`
fallback, so construction order no longer matters.

**Fixed — `explain_plan()` / `analyze_plan()` described a different
query than the one that ran.** Both built their children straight from
`self._inner`, bypassing the limit/offset rewrite in `to_batches`, and
reported `skip=2, fetch=2` while execution used `skip=0, fetch=4`. Child
preparation now lives in one `_create_child_queries()` helper used by
all three.

> **Visible change to `explain_plan()` output:** because the plan is now
built from the real execution children, which carry `with_row_id()`, the
two `ProjectionExec` lines gain a `_rowid` column. The doctest is
updated to match. This is the diagnostic becoming truthful rather than
the assertion being weakened — it is still an exact-match comparison.

**Not fixed here — RRF candidate-pool invariance.** Widening each
sub-query to `limit + offset` does change the candidate pool between
page requests, so the fused ranking can shift and pagination can still
repeat rows. That's a real problem, but it is exactly what the merged
sync path does today:

```python
# LanceHybridQueryBuilder (sync), merged in #3769
sub_query_limit = self._limit + (self._offset or 0)
```

Making the pool invariant means choosing a contract — a fixed candidate
pool, or an explicit cursor — and that ought to apply to sync and async
together rather than letting the two paths diverge. I've asked in the
review thread which way you'd prefer, and I'm happy to do it here or in
a follow-up covering both paths.

So, to be precise about what this PR delivers: it makes `.offset()` take
effect on the async hybrid path and makes the diagnostics honest. It
does not make hybrid pagination stable across pages under reranking —
that needs the contract decision above.

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.