diff --git a/docs/docs/pypaimon/multimodal-search.md b/docs/docs/pypaimon/multimodal-search.md
index 8c98e2e3313e..c7a3c7d0893e 100644
--- a/docs/docs/pypaimon/multimodal-search.md
+++ b/docs/docs/pypaimon/multimodal-search.md
@@ -60,8 +60,18 @@ has exactly one text column. To target a specific text column, pass `column`.
 
 Use `pre_filter` to prune search candidates before ranking. Use `where()` to
 filter the rows read from the search result. Both `pre_filter` and `where()`
-accept SQL-like predicate strings. For full-text search, `pre_filter` must only
-reference partition columns.
+accept SQL-like predicate strings. On data-evolution tables, full-text
+`pre_filter` supports ordinary data columns as well as partition columns.
+For primary-key full-text search, only partition predicates are supported.
+
+Full-text data predicates are evaluated before Top-K selection. PyPaimon reads
+the predicate's columns and row IDs at the search snapshot, then passes the
+matching row IDs to the full-text index. Scalar indexes can prune this read,
+but partial scalar-index coverage does not exclude matching rows covered by
+the full-text search plan. This filter read can scan all candidate rows;
+partition-only predicates keep the existing partition-pruning path. For
+unindexed data in `full-text-index.search-mode=full`, filtering preserves the
+unfiltered corpus used to calculate BM25 statistics.
 
 For data-evolution vector search, a scalar index may return candidates rather than exact
 matches, for example for BTree string-prefix or substring predicates, or when
@@ -92,7 +102,7 @@ neighbors = (
 )
 
 matches = (
-    docs.search("paimon vector", column="content")
+    docs.search("paimon vector", column="content", pre_filter="category = 'lake'")
     .limit(10)
     .to_pandas()
 )
@@ -221,8 +231,11 @@ text column when the table has exactly one text column. To target a specific
 text column, pass `column` to `pm.text_route`.
 
 `pre_filter` is applied before ranking. It accepts a SQL-like predicate string.
-When a hybrid query has a full-text route, `pre_filter` must only reference
-partition columns.
+On data-evolution tables, ordinary data predicates are applied to both vector
+and full-text routes before each route selects its candidates. The vector
+route retains the `global-index.filter.refine-from-data` behavior described
+above, while the full-text route verifies data predicates through a filter-column
+read. Partition-only predicates prune both routes without that extra read.
 
 ```python
 # This example assumes the table is partitioned by dt.
diff --git a/paimon-python/pypaimon/multimodal/query.py b/paimon-python/pypaimon/multimodal/query.py
index af2a983f9a06..c3b839a4e6e5 100644
--- a/paimon-python/pypaimon/multimodal/query.py
+++ b/paimon-python/pypaimon/multimodal/query.py
@@ -599,7 +599,7 @@ def _execute_fts(self, query):
             .with_limit(limit)
         )
         if query._pre_filter is not None:
-            builder = builder.with_partition_filter(query._pre_filter)
+            builder = builder.with_filter(query._pre_filter)
         return builder.execute_local()
 
 
diff --git a/paimon-python/pypaimon/table/source/full_text_read.py b/paimon-python/pypaimon/table/source/full_text_read.py
index cccd1a90e1dc..c4236b664628 100644
--- a/paimon-python/pypaimon/table/source/full_text_read.py
+++ b/paimon-python/pypaimon/table/source/full_text_read.py
@@ -39,6 +39,7 @@
 )
 from pypaimon.table.source.full_text_scan import FullTextScanPlan
 from pypaimon.utils.range import Range
+from pypaimon.utils.roaring_bitmap import RoaringBitmap64
 
 
 class FullTextRead(ABC):
@@ -62,6 +63,7 @@ def __init__(
         text_column,
         query: str,
         partition_filter=None,
+        filter_=None,
     ):
         self._table = table
         self._limit = limit
@@ -72,6 +74,7 @@ def __init__(
                 % self._text_columns)
         self._query = query
         self._partition_filter = partition_filter
+        self._filter = filter_
 
     def read_plan(self, plan: FullTextScanPlan) -> GlobalIndexResult:
         reader = copy(self)
@@ -88,9 +91,25 @@ def read(self, splits: List[FullTextSearchSplit]) -> GlobalIndexResult:
             splits_by_column.setdefault(split.column_name, []).append(split)
         live_rows = global_index_live_row_filter.live_rows(
             self._table, self._partition_filter)
+        if self._filter is not None:
+            from pypaimon.table.source.global_index_row_filter import matching_rows
+
+            candidates = RoaringBitmap64()
+            for split in index_splits:
+                candidates.add_range(split.row_range_start, split.row_range_end)
+            for row_range in _raw_row_ranges(raw_splits):
+                candidates.add_range(row_range.from_, row_range.to)
+            # Scalar indexes may prune this read, but incomplete scalar coverage
+            # must not exclude rows covered by the full-text search plan.
+            table = self._table.copy({"scalar-index.search-mode": "full"})
+            matched = matching_rows(table, self._filter, candidates, self._partition_filter)
+            live_rows = matched if live_rows is None else RoaringBitmap64.and_(live_rows, matched)
         indexed_result = self._eval_column_query(splits_by_column, live_rows)
-        raw_result = self._read_raw_search(
-            _raw_row_ranges(raw_splits), _index_type(index_splits))
+        raw_ranges = _raw_row_ranges(raw_splits)
+        if self._filter is None:
+            raw_result = self._read_raw_search(raw_ranges, _index_type(index_splits))
+        else:
+            raw_result = self._read_raw_search(raw_ranges, _index_type(index_splits), live_rows)
         return indexed_result.or_(raw_result).top_k(self._limit)
 
     def _eval_column_query(
@@ -170,7 +189,7 @@ def _eval(self, row_range_start, row_range_end, full_text_index_files,
         future.add_done_callback(lambda _: reader.close())
         return future
 
-    def _read_raw_search(self, raw_row_ranges, index_type):
+    def _read_raw_search(self, raw_row_ranges, index_type, include_row_ids=None):
         raw_row_ranges = Range.sort_and_merge_overlap(raw_row_ranges, True)
         if not raw_row_ranges:
             return DictBasedScoredIndexResult({})
@@ -179,6 +198,15 @@ def _read_raw_search(self, raw_row_ranges, index_type):
 
         row_range_start = raw_row_ranges[0].from_
         row_range_end = raw_row_ranges[-1].to
+        search_kwargs = {}
+        if include_row_ids is not None:
+            raw_rows = GlobalIndexResult.from_ranges(raw_row_ranges).results()
+            include_row_ids = RoaringBitmap64.and_(include_row_ids, raw_rows)
+            if include_row_ids.is_empty():
+                return DictBasedScoredIndexResult({})
+            search_kwargs["include_row_ids"] = include_row_ids
+        # Build the same corpus as an unfiltered query so BM25 statistics stay
+        # unchanged. Filter row IDs in the native search, before selecting hits.
         table = self._read_raw_rows(raw_row_ranges)
         if table is None or table.num_rows == 0:
             return DictBasedScoredIndexResult({})
@@ -196,6 +224,7 @@ def _read_raw_search(self, raw_row_ranges, index_type):
             row_range_start,
             self._query,
             _candidate_limit(row_range_start, row_range_end),
+            **search_kwargs,
         ).top_k(self._limit)
 
     def _read_raw_rows(self, raw_row_ranges):
@@ -300,7 +329,7 @@ def _index_type(index_splits):
     return None
 
 
-def _search_raw_full_text(index_bytes, row_range_start, query, limit):
+def _search_raw_full_text(index_bytes, row_range_start, query, limit, include_row_ids=None):
     from paimon_ftindex import FullTextIndexReader
     from pypaimon.globalindex.full_text.native_full_text_global_index_reader import (
         PaimonFullTextInput,
@@ -308,7 +337,11 @@ def _search_raw_full_text(index_bytes, row_range_start, query, limit):
 
     reader = FullTextIndexReader(PaimonFullTextInput(BytesIO(index_bytes)))
     try:
-        row_ids, scores = reader.search(query, limit=limit)
+        search_kwargs = {}
+        if include_row_ids is not None:
+            relative_ids = GlobalIndexResult.create(include_row_ids).offset(-row_range_start).results()
+            search_kwargs["filter_bytes"] = relative_ids.serialize()
+        row_ids, scores = reader.search(query, limit=limit, **search_kwargs)
         return DictBasedScoredIndexResult(
             {
                 row_range_start + row_id: score
diff --git a/paimon-python/pypaimon/table/source/full_text_scan.py b/paimon-python/pypaimon/table/source/full_text_scan.py
index 29b44ee5b0d4..ae5d089d0bc8 100644
--- a/paimon-python/pypaimon/table/source/full_text_scan.py
+++ b/paimon-python/pypaimon/table/source/full_text_scan.py
@@ -115,18 +115,17 @@ def index_file_filter(entry):
                     IndexFullTextSearchSplit(
                         column_name, range_key.from_, range_key.to, files))
 
-        if all_index_files:
-            raw_row_ranges = DataEvolutionGlobalIndexCoverage(
-                self._table,
-                snapshot,
-                partition_filter,
-                all_index_files,
-            ).unindexed_ranges(
-                list(text_column_ids),
-                search_mode=self._table.options.full_text_index_search_mode(),
-            )
-            if raw_row_ranges:
-                splits.append(RawFullTextSearchSplit(raw_row_ranges))
+        raw_row_ranges = DataEvolutionGlobalIndexCoverage(
+            self._table,
+            snapshot,
+            partition_filter,
+            all_index_files,
+        ).unindexed_ranges(
+            list(text_column_ids),
+            search_mode=self._table.options.full_text_index_search_mode(),
+        )
+        if raw_row_ranges:
+            splits.append(RawFullTextSearchSplit(raw_row_ranges))
 
         return FullTextScanPlan(splits, snapshot)
 
diff --git a/paimon-python/pypaimon/table/source/full_text_search_builder.py b/paimon-python/pypaimon/table/source/full_text_search_builder.py
index 1e094e543f62..764bb0943ea3 100644
--- a/paimon-python/pypaimon/table/source/full_text_search_builder.py
+++ b/paimon-python/pypaimon/table/source/full_text_search_builder.py
@@ -48,6 +48,11 @@ def with_partition_filter(self, partition_filter) -> 'FullTextSearchBuilder':
         """Partition predicate used to prune index manifest entries."""
         pass
 
+    @abstractmethod
+    def with_filter(self, predicate) -> 'FullTextSearchBuilder':
+        """Filter data-evolution rows before full-text ranking."""
+        pass
+
     @abstractmethod
     def new_full_text_scan(self) -> FullTextScan:
         """Create full-text scan to scan index files."""
@@ -72,6 +77,7 @@ def __init__(self, table: 'FileStoreTable'):
         self._field_name: Optional[str] = None
         self._query: Optional[str] = None
         self._partition_filter = None
+        self._filter = None
 
     def with_limit(self, limit: int) -> 'FullTextSearchBuilder':
         self._limit = limit
@@ -82,6 +88,20 @@ def with_query(self, field_name: str, query: str) -> 'FullTextSearchBuilder':
         self._query = query
         return self
 
+    def with_filter(self, predicate) -> 'FullTextSearchBuilder':
+        if predicate is None:
+            return self
+        from pypaimon.read.push_down_utils import _split_and, _get_all_fields
+
+        partition_keys = set(self._table.partition_keys or [])
+        for part in _split_and(predicate):
+            if partition_keys and _get_all_fields(part).issubset(partition_keys):
+                self.with_partition_filter(part)
+            else:
+                self._filter = (part if self._filter is None else
+                                PredicateBuilder.and_predicates([self._filter, part]))
+        return self
+
     def with_partition_filter(self, partition_filter) -> 'FullTextSearchBuilder':
         if partition_filter is None:
             self._partition_filter = None
@@ -132,6 +152,8 @@ def new_full_text_scan(self) -> FullTextScan:
     def new_full_text_read(self) -> FullTextRead:
         if self._limit <= 0:
             raise ValueError("Limit must be positive, set via with_limit()")
+        if self._filter is not None and not self._table.options.data_evolution_enabled():
+            raise NotImplementedError("Full-text row filters require a data-evolution table.")
         definition = self._primary_key_full_text_definition()
         if definition is not None:
             from pypaimon.common.options.core_options import GlobalIndexSearchMode
@@ -153,6 +175,7 @@ def new_full_text_read(self) -> FullTextRead:
             self._text_columns(),
             self._query,
             partition_filter=self._partition_filter,
+            filter_=self._filter,
         )
 
     def _text_columns(self):
diff --git a/paimon-python/pypaimon/table/source/global_index_row_filter.py b/paimon-python/pypaimon/table/source/global_index_row_filter.py
new file mode 100644
index 000000000000..0dc087f56a97
--- /dev/null
+++ b/paimon-python/pypaimon/table/source/global_index_row_filter.py
@@ -0,0 +1,42 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Exact row filtering for data-evolution search candidates."""
+
+from pypaimon.read.table_read import _ClosableArrowBatchReader
+from pypaimon.table.special_fields import SpecialFields
+from pypaimon.table.source.global_index_live_row_filter import table_at_snapshot
+from pypaimon.utils.roaring_bitmap import RoaringBitmap64
+
+
+def matching_rows(table, predicate, candidates, partition_filter=None, snapshot=None):
+    """Read only filter dependencies and row IDs, closing both reader and source."""
+    matched = RoaringBitmap64()
+    if candidates.is_empty():
+        return matched
+    table = table_at_snapshot(table, snapshot)
+    builder = (table.new_read_builder().with_filter(predicate)
+               .with_projection([SpecialFields.ROW_ID.name]))
+    if partition_filter is not None:
+        builder = builder.with_partition_filter(partition_filter)
+    splits = builder.new_scan().with_row_ranges(candidates.to_range_list()).plan().splits()
+    reader, batches = builder.new_read()._new_arrow_batch_reader(splits)
+    with _ClosableArrowBatchReader(reader, batches) as batch_reader:
+        for batch in batch_reader:
+            for row_id in batch.column(SpecialFields.ROW_ID.name).to_pylist():
+                matched.add(row_id)
+    return matched
diff --git a/paimon-python/pypaimon/table/source/hybrid_search_builder.py b/paimon-python/pypaimon/table/source/hybrid_search_builder.py
index a3e43c3fb6cc..03fe4471c120 100644
--- a/paimon-python/pypaimon/table/source/hybrid_search_builder.py
+++ b/paimon-python/pypaimon/table/source/hybrid_search_builder.py
@@ -361,13 +361,6 @@ def _validate_search(self):
             raise ValueError("Routes cannot be empty")
         if self._limit <= 0:
             raise ValueError("Limit must be positive, got: %s" % self._limit)
-        if self._filter is not None:
-            for route in self._routes:
-                if route.is_full_text():
-                    raise ValueError(
-                        "Hybrid search with full-text routes does not support "
-                        "non-partition filters because full-text indexes cannot "
-                        "apply row-id pre-filters before top-k ranking.")
 
     def _new_vector_search_builder(self, route):
         builder = (
@@ -391,6 +384,8 @@ def _new_full_text_search_builder(self, route):
         )
         if self._partition_filter is not None:
             builder.with_partition_filter(self._partition_filter)
+        if self._filter is not None:
+            builder.with_filter(self._filter)
         return builder
 
     def _rrf(self, route_results):
diff --git a/paimon-python/pypaimon/table/source/vector_search_read.py b/paimon-python/pypaimon/table/source/vector_search_read.py
index 59909966d0f5..0df1fec9839b 100644
--- a/paimon-python/pypaimon/table/source/vector_search_read.py
+++ b/paimon-python/pypaimon/table/source/vector_search_read.py
@@ -202,23 +202,9 @@ def _scalar_matched_rows(self, splits, snapshot=None):
         return self._matching_candidate_rows(candidates, snapshot)
 
     def _matching_candidate_rows(self, candidates, snapshot):
-        from pypaimon.read.table_read import _ClosableArrowBatchReader
+        from pypaimon.table.source.global_index_row_filter import matching_rows
 
-        matched = RoaringBitmap64()
-        if candidates.is_empty():
-            return matched
-        table = global_index_live_row_filter.table_at_snapshot(self._table, snapshot)
-        builder = (table.new_read_builder().with_filter(self._filter)
-                   .with_projection([SpecialFields.ROW_ID.name]))
-        if self._partition_filter is not None:
-            builder = builder.with_partition_filter(self._partition_filter)
-        splits = builder.new_scan().with_row_ranges(candidates.to_range_list()).plan().splits()
-        reader, batches = builder.new_read()._new_arrow_batch_reader(splits)
-        with _ClosableArrowBatchReader(reader, batches) as batch_reader:
-            for batch in batch_reader:
-                for row_id in batch.column(SpecialFields.ROW_ID.name).to_pylist():
-                    matched.add(row_id)
-        return matched
+        return matching_rows(self._table, self._filter, candidates, self._partition_filter, snapshot)
 
     def _pre_filter(self, splits, snapshot=None):
         # Backwards-compatible helper used by older tests/callers.
diff --git a/paimon-python/pypaimon/tests/full_text_scalar_filter_test.py b/paimon-python/pypaimon/tests/full_text_scalar_filter_test.py
new file mode 100644
index 000000000000..523f573dea54
--- /dev/null
+++ b/paimon-python/pypaimon/tests/full_text_scalar_filter_test.py
@@ -0,0 +1,179 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from unittest.mock import patch
+
+import pyarrow as pa
+import pytest
+
+import pypaimon.multimodal as pm
+from pypaimon.table.source.full_text_read import DataEvolutionFullTextRead
+from pypaimon.table.source.global_index_row_filter import matching_rows
+
+
+@pytest.fixture
+def docs(tmp_path):
+    pytest.importorskip("paimon_ftindex")
+    schema = pa.schema([("id", pa.int64()), ("text", pa.string()), ("label", pa.string()),
+                        ("pt", pa.string()), ("embedding", pa.list_(pa.float32(), 2))])
+    table = pm.connect(options={"warehouse": str(tmp_path)}).create_table(
+        "docs", schema=schema, partitioned=["pt"], options={
+            "file.format": "parquet", "vector.file.format": "parquet", "read.batch-size": "1",
+            "full-text-index.search-mode": "full", "vector-index.search-mode": "full"})
+    table.add(pa.table({"id": [0, 1, 2], "text": ["paimon", "paimon paimon", "paimon long document"],
+                        "label": ["other", "other", "target"], "pt": ["a"] * 3,
+                        "embedding": [[0., 1.], [1., 1.], [2., 1.]]}, schema=schema))
+    table.raw_table.copy({"deletion-vectors.enabled": "false"}).create_global_index("text", "full-text")
+    return table
+
+
+def append_rows(docs):
+    docs.add(pa.table({"id": [3, 4], "text": ["paimon", "paimon another long document"],
+                       "label": ["other", "target"], "pt": ["b"] * 2,
+                       "embedding": [[3., 1.], [4., 1.]]}, schema=docs.scan().to_arrow().schema))
+
+
+def builder(docs, predicate=None, limit=10):
+    result = docs.raw_table.new_full_text_search_builder().with_query(
+        "text", '{"match":{"query":"paimon"}}').with_limit(limit)
+    if predicate is not None:
+        from pypaimon.common.where_parser import parse_where_clause
+        result.with_filter(parse_where_clause(predicate, docs.raw_table.fields))
+    return result
+
+
+def scores(result):
+    return {row_id: result.score_getter()(row_id) for row_id in result.results()}
+
+
+@pytest.mark.parametrize("kind", [None, "btree", "bitmap"])
+@pytest.mark.parametrize("raw", [False, True])
+def test_full_text_data_filters_precede_top_k_and_preserve_scores(docs, kind, raw):
+    if kind:
+        docs.raw_table.copy({"deletion-vectors.enabled": "false"}).create_global_index("label", kind)
+    if raw:
+        append_rows(docs)
+    expected = scores(builder(docs).execute_local())
+    actual = scores(builder(docs, "label LIKE '%target%'", limit=1).execute_local())
+    candidates = [2, 4] if raw else [2]
+    best = max(candidates, key=lambda row_id: (expected[row_id], -row_id))
+    assert actual == {best: expected[best]}
+    public = docs.search("paimon", column="text", pre_filter="label LIKE '%target%'").select(["id"]).limit(1)
+    assert public.to_list() == [{"id": best}]
+
+
+def test_partial_scalar_coverage_does_not_drop_full_text_matches(docs):
+    docs.raw_table.copy({"deletion-vectors.enabled": "false"}).create_global_index("label", "btree")
+    append_rows(docs)
+    docs.raw_table.copy({"deletion-vectors.enabled": "false"}).create_global_index("text", "full-text")
+    docs.raw_table = docs.raw_table.copy({"scalar-index.search-mode": "fast"})
+    assert docs.search("paimon", column="text", pre_filter="label = 'target'").select(["id"]).limit(10).to_list() == [
+        {"id": 2}, {"id": 4}]
+
+
+@pytest.mark.parametrize("predicate, expected", [
+    ("pt = 'b' AND label = 'target'", [4]),
+    ("pt = 'a' OR id = 4", [0, 1, 2, 4]),
+    ("id = 2", [2]),
+])
+def test_partition_and_data_predicates_keep_boolean_semantics(docs, predicate, expected):
+    append_rows(docs)
+    actual = docs.search("paimon", column="text", pre_filter=predicate).select(["id"]).limit(10).to_list()
+    assert sorted(row["id"] for row in actual) == expected
+
+
+def test_partition_only_filter_keeps_existing_path(docs):
+    with patch("pypaimon.table.source.global_index_row_filter.matching_rows",
+               side_effect=AssertionError("partition-only data read")):
+        assert len(docs.search("paimon", column="text", pre_filter="pt = 'a'").limit(10).to_list()) == 3
+
+
+@pytest.mark.parametrize("mode, expected", [("fast", []), ("full", [{"id": 4}])])
+def test_partition_without_full_text_index_respects_search_mode(docs, mode, expected):
+    append_rows(docs)
+    docs.raw_table = docs.raw_table.copy({"full-text-index.search-mode": mode})
+    search = docs.search("paimon", column="text", pre_filter="pt = 'b' AND label = 'target'")
+    assert search.select(["id"]).limit(1).to_list() == expected
+
+
+def test_non_partitioned_table_filters_before_top_k(tmp_path):
+    pytest.importorskip("paimon_ftindex")
+    schema = pa.schema([("id", pa.int64()), ("text", pa.string())])
+    docs = pm.connect(options={"warehouse": str(tmp_path)}).create_table(
+        "docs", schema=schema, options={"file.format": "parquet"})
+    docs.add(pa.table({"id": [0, 1], "text": ["paimon", "paimon long document"]}, schema=schema))
+    docs.raw_table.copy({"deletion-vectors.enabled": "false"}).create_global_index("text", "full-text")
+    assert docs.search("paimon", column="text").where("id = 1").limit(1).to_list() == []
+    assert docs.search("paimon", column="text", pre_filter="id = 1").select(["id"]).limit(1).to_list() == [
+        {"id": 1}]
+
+
+def test_empty_matches_skip_native_search(docs):
+    with patch.object(DataEvolutionFullTextRead, "_eval", side_effect=AssertionError("empty search")):
+        assert docs.search("paimon", column="text", pre_filter="id < 0").limit(1).to_list() == []
+
+
+@pytest.mark.parametrize("selector", ["snapshot", "tag"])
+def test_filters_use_historical_snapshot_and_deletions(docs, selector):
+    saved = docs.raw_table.snapshot_manager().get_latest_snapshot()
+    docs.raw_table.create_tag("saved", snapshot_id=saved.id)
+    options = {"snapshot_id": saved.id} if selector == "snapshot" else {"tag_name": "saved"}
+    docs.delete("id = 2")
+    search = docs.search("paimon", column="text", pre_filter="label = 'target'", **options)
+    assert search.select(["id"]).limit(1).to_list() == [{"id": 2}]
+    assert docs.search("paimon", column="text", pre_filter="label = 'target'").limit(1).to_list() == []
+
+
+def test_commit_during_filter_read_keeps_execution_snapshot(docs):
+    search = docs.search("paimon", column="text", pre_filter="label = 'target'").select(["id"]).limit(1)
+
+    def verify(table, predicate, candidates, partition_filter=None, snapshot=None):
+        docs.update("id = 2", {"label": "changed"})
+        return matching_rows(table, predicate, candidates, partition_filter, snapshot)
+
+    with patch("pypaimon.table.source.global_index_row_filter.matching_rows", verify):
+        assert search.to_list() == [{"id": 2}]
+    assert search.to_list() == []
+
+
+def test_hybrid_applies_data_filter_to_both_routes(docs):
+    search = docs.search_hybrid([
+        pm.vector_route("embedding", [0., 1.], limit=1),
+        pm.text_route("paimon", column="text", limit=1),
+    ], pre_filter="label = 'target'").select(["id"]).limit(1)
+    assert search.to_list() == [{"id": 2}]
+
+
+def test_data_filter_does_not_project_vectors_or_text(docs):
+    from pypaimon.read.table_read import TableRead
+    original = TableRead._new_arrow_batch_reader
+    projections = []
+
+    def read(reader, *args, **kwargs):
+        projections.append([field.name for field in reader.read_type])
+        return original(reader, *args, **kwargs)
+
+    with patch.object(TableRead, "_new_arrow_batch_reader", read):
+        assert docs.search("paimon", column="text", pre_filter="label = 'target'").select(["id"]).limit(1).to_list()
+    assert projections == [["_ROW_ID"]]
+    assert all("embedding" not in fields and "text" not in fields for fields in projections)
+
+
+def test_primary_key_row_filter_is_explicitly_unsupported(docs):
+    docs.raw_table = docs.raw_table.copy({"data-evolution.enabled": "false"})
+    with pytest.raises(NotImplementedError, match="data-evolution"):
+        builder(docs, "id = 2").new_full_text_read()
diff --git a/paimon-python/pypaimon/tests/vector_search_filter_test.py b/paimon-python/pypaimon/tests/vector_search_filter_test.py
index 1d2731666f21..837e792ea7fa 100644
--- a/paimon-python/pypaimon/tests/vector_search_filter_test.py
+++ b/paimon-python/pypaimon/tests/vector_search_filter_test.py
@@ -2683,7 +2683,7 @@ def new_full_text_search_builder():
         self.assertEqual(("query", "content", match_query("paimon search", "And")),
                          captured_builders[1].calls[0])
 
-    def test_hybrid_search_rejects_data_filter_with_full_text_route(self):
+    def test_hybrid_search_forwards_data_filter_to_full_text_route(self):
         from pypaimon.table.source.hybrid_search_builder import (
             HybridSearchBuilderImpl,
         )
@@ -2701,9 +2701,8 @@ def test_hybrid_search_rejects_data_filter_with_full_text_route(self):
             .with_limit(5)
         )
 
-        with self.assertRaises(ValueError) as ctx:
-            builder.route_builders()
-        self.assertIn("full-text routes", str(ctx.exception))
+        route = builder.route_builders()[0]
+        self.assertEqual(pb.equal("id", 1), route.search_builder._filter)
 
     def test_hybrid_search_rejects_full_text_route_options(self):
         from pypaimon.table.source.hybrid_search_builder import (