[python] Add explain and profiling for local search

docs/docs/pypaimon/multimodal-search.md

@@ -63,6 +63,14 @@ 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.
 
+Each execution of `search`, `search_vectors`, or `search_hybrid` reads one
+snapshot across candidate search, filtering, reranking, and result lookup.
+Concurrent commits become visible on the next execution, including when reusing
+the same query object. Explicit snapshot, tag, and timestamp selectors are
+honored. All routes in a hybrid search and all vectors in a batch share that
+execution's snapshot. Keep the snapshot's data files available for the duration
+of the query; capturing a read view does not prevent snapshot expiration.
+
 ```python
 neighbors = (
     docs.search(
@@ -193,3 +201,88 @@ batch_neighbors = (
     .to_list()
 )
 ```
+
+## Explain and Profile Search
+
+Call `explain()` on a vector, batch vector, full-text, or hybrid query to inspect
+its planned index and raw-data work. It returns a `SearchExplainResult` with
+structured fields and a printable summary. It does not run vector/text searches
+or fetch result rows. Planning can read data to evaluate primary-key scalar
+predicates, so it is not always metadata-only.
+
+```python
+query = docs.search(query_vector, column="embedding").select(["content"]).limit(10)
+plan = query.explain()
+print(plan)
+
+route = plan.routes[0]
+print(route.index_file_count, route.index_bytes, route.index_types)
+print(route.indexed_range_rows, route.raw_range_rows, route.overlapping_range_rows)
+```
+
+Each route reports its search mode, column, candidate limit, query count,
+planning snapshot, index splits/files/bytes, raw splits, scalar-index file count,
+and whether scalar or partition filters are configured.
+Hybrid plans retain route order, weights, the fusion
+ranker, and the route worker limit. A query's projection and post-filter flag
+are included; `where()` applies during lookup after search top-k and can reduce
+the number of returned rows.
+
+Coverage counts are unions of inclusive row-ID ranges in the plan, not live-row
+counts or measured ANN recall. Indexed and raw ranges can overlap, for example
+when scalar-index coverage requires fallback. Do not add these counts as if they
+were disjoint. Primary-key source-file plans report `None` for global row-ID
+coverage because their positions are local to source files.
+
+Call `profile()` to execute the search once and obtain its result together with
+runtime measurements. Use `profile.result` directly; subsequently calling
+`to_arrow()` would execute a second search.
+
+```python
+profile = query.profile()
+print(profile)
+neighbors = profile.result  # Arrow table, with the normal projection and filters
+print(profile.elapsed_ms, profile.lookup_ms, profile.output_rows)
+print(profile.route_metrics[0]["timings_ms"])
+print(profile.route_metrics[0]["counters"])
+
+batch_profile = docs.search_vectors(query_vectors, column="embedding").limit(10).profile()
+batch_neighbors = batch_profile.result  # One Arrow table per query vector
+```
+
+`SearchProfileResult.plan` describes that execution's plan. A separate earlier
+`explain()` may describe a different snapshot if the table changes. Each
+`profile()` execution shares one snapshot across hybrid routes, filtering,
+raw fallback, and result lookup, just like normal search. Each route's
+`snapshot_id` and the entries in `lookup_snapshot_ids` describe that same read
+view, with one lookup entry per batch query. Reusing the query captures a fresh
+snapshot unless it has an explicit time-travel selector.
+
+The built-in local vector, batch vector, full-text, and hybrid search builders
+also expose `explain()` and `profile()`. Builder profiles return their normal
+scored index result (a list for batch search), without fetching Arrow rows, so
+`lookup_ms`, `lookup_snapshot_ids`, and `output_rows` are `None`.
+
+Runtime metrics are opt-in. Normal execution does not collect profiling clocks
+or counters. The available measurements are:
+
+| Measurement | Meaning |
+| --- | --- |
+| `elapsed_ms` | Wall time for this call, including lookup for multimodal queries. |
+| `planning`, `search` | Per-route planning and search wall time in `timings_ms`. |
+| `index_open`, `index_search` | Reader opening and accumulated index-search time. Full-text lazy loading is included in `index_search`. |
+| `pre_filter`, `raw_read_score`, `refine` | Instrumented filtering, raw-data reading/scoring, and vector refinement time. Single-vector refinement also includes its nested `raw_read_score` stage. |
+| `fusion_ms`, `lookup_ms` | Hybrid result fusion and multimodal result lookup time. |
+| `index_searches`, `peak_index_searches` | Number of index calls and peak outstanding calls per route, not native worker-thread counts. A batch call counts once. |
+| `index_rows_before_filter`, `index_rows_after_filter` | Sum of submitted index row-range sizes before and after include-row-ID filtering; fully pruned splits make no index call. These are not counts of vectors visited by ANN. |
+| `index_candidates`, `refine_candidates` | Candidates returned by indexes or submitted for refinement, summed across shards/queries before final merging. |
+| `raw_rows_read`, `refine_rows_read` | Rows yielded to raw scoring or refinement. Batch refinement reads shared candidate rows once; these are not storage-level I/O row counts. |
+| `result_rows`, `output_rows` | Per-route result count before fusion/lookup, and final Arrow row count. Batch counts sum all queries. |
+
+Times are milliseconds. Stage times are inclusive and may overlap or sum across
+concurrent work; adding them does not yield wall time. Missing measurements mean
+the phase was not instrumented or invoked, not necessarily that it was free.
+Primary-key readers expose total planning/search time and inherited index/refine
+stages, but do not currently report every raw-read or refinement row counter.
+Profiling incurs the real search cost plus measurement overhead; it is not an
+estimate or a benchmark isolated from caches and other queries.

paimon-python/pypaimon/globalindex/data_evolution_global_index_scanner.py

@@ -344,6 +344,8 @@ def close(self):
 def _resolve_snapshot(table, snapshot):
     if snapshot is not None:
         return snapshot
+    if hasattr(table, "_read_snapshot"):
+        return table._read_snapshot
     snapshot_manager = table.snapshot_manager()
     if snapshot_manager is None:
         return None

paimon-python/pypaimon/multimodal/query.py

@@ -15,6 +15,7 @@
 # specific language governing permissions and limitations
 # under the License.
 
+from copy import copy
 from typing import Any, Callable, Dict, List, Optional, Tuple
 
 import pyarrow as pa
@@ -109,10 +110,12 @@ def _effective_projection(self):
             projection.append(SpecialFields.ROW_ID.name)
         return projection
 
-    def _read_global_index_result(self, result):
+    def _read_global_index_result(self, result, snapshot_ids=None):
         read_builder = self._configured_read_builder()
         scan = read_builder.new_scan().with_global_index_result(result)
         plan = scan.plan()
+        if snapshot_ids is not None:
+            snapshot_ids.append(plan.snapshot_id)
         return read_builder.new_read().to_arrow(plan.splits())
 
     def to_pandas(self):
@@ -446,6 +449,47 @@ def _and_predicate(self, left, right):
 
 class _PreFilterQuery(ScanQuery):
 
+    def explain(self):
+        """Plan this search without executing it or fetching result rows."""
+        plan = self._for_execution()._search_builder().explain()
+        plan.projection = self._effective_projection()
+        plan.has_post_filter = self._predicate is not None
+        return plan
+
+    def profile(self):
+        """Run this search once, returning Arrow results and execution metrics."""
+        import time
+
+        start = time.perf_counter()
+        query = self._for_execution()
+        profile = query._search_builder().profile()
+        profile.plan.projection = self._effective_projection()
+        profile.plan.has_post_filter = self._predicate is not None
+        lookup_start = time.perf_counter()
+        profile.lookup_snapshot_ids = []
+        if isinstance(profile.result, list):
+            profile.result = [query._read_global_index_result(result, profile.lookup_snapshot_ids)
+                              for result in profile.result]
+            profile.output_rows = sum(table.num_rows for table in profile.result)
+        else:
+            profile.result = query._read_global_index_result(profile.result, profile.lookup_snapshot_ids)
+            profile.output_rows = profile.result.num_rows
+        profile.lookup_ms = (time.perf_counter() - lookup_start) * 1000
+        profile.elapsed_ms = (time.perf_counter() - start) * 1000
+        return profile
+
+    def _search_builder(self):
+        raise NotImplementedError
+
+    def _for_execution(self):
+        from pypaimon.snapshot.time_travel_util import TimeTravelUtil
+        query = copy(self)
+        query._table = self._table._copy_with_snapshot(TimeTravelUtil.resolve_snapshot(self._table))
+        return query
+
+    def to_arrow(self):
+        return ScanQuery.to_arrow(self._for_execution())
+
     def __init__(
             self,
             table,
@@ -502,17 +546,20 @@ def __init__(
             table, result_factory=self._execute_vector, pre_filter=pre_filter)
 
     def _execute_vector(self, query):
-        limit = query._limit if query._limit is not None else 10
+        return query._search_builder().execute_local()
+
+    def _search_builder(self):
+        limit = self._limit if self._limit is not None else 10
         builder = (
             self._table.new_vector_search_builder()
             .with_vector_column(self._vector_column)
             .with_query_vector(self._vector)
             .with_limit(limit)
             .with_options(self._vector_options)
         )
-        if query._pre_filter is not None:
-            builder = builder.with_filter(query._pre_filter)
-        return builder.execute_local()
+        if self._pre_filter is not None:
+            builder = builder.with_filter(self._pre_filter)
+        return builder
 
 
 class TextQuery(_PreFilterQuery):
@@ -524,15 +571,18 @@ def __init__(self, table, text_query, pre_filter=None):
             table, result_factory=self._execute_fts, pre_filter=pre_filter)
 
     def _execute_fts(self, query):
-        limit = query._limit if query._limit is not None else 10
+        return query._search_builder().execute_local()
+
+    def _search_builder(self):
+        limit = self._limit if self._limit is not None else 10
         builder = (
             self._table.new_full_text_search_builder()
             .with_query(self._text_query["column"], self._text_query["query"])
             .with_limit(limit)
         )
-        if query._pre_filter is not None:
-            builder = builder.with_partition_filter(query._pre_filter)
-        return builder.execute_local()
+        if self._pre_filter is not None:
+            builder = builder.with_partition_filter(self._pre_filter)
+        return builder
 
 
 class HybridQuery(_PreFilterQuery):
@@ -558,7 +608,10 @@ def rerank(self, ranker):
         return self
 
     def _execute_hybrid(self, query):
-        final_limit = query._limit if query._limit is not None else 10
+        return query._search_builder().execute_local()
+
+    def _search_builder(self):
+        final_limit = self._limit if self._limit is not None else 10
         route_limit = self._route_limit or final_limit
         builder = (
             self._table.new_hybrid_search_builder()
@@ -581,9 +634,9 @@ def _execute_hybrid(self, query):
                 weight=route["weight"],
                 options=route["options"],
             )
-        if query._pre_filter is not None:
-            builder = builder.with_filter(query._pre_filter)
-        return builder.execute_local()
+        if self._pre_filter is not None:
+            builder = builder.with_filter(self._pre_filter)
+        return builder
 
 
 class BatchVectorQuery(_PreFilterQuery):
@@ -602,9 +655,10 @@ def __init__(
         super().__init__(table, pre_filter=pre_filter)
 
     def to_arrow(self):
+        query = self._for_execution()
         return [
-            self._read_global_index_result(result)
-            for result in self._execute_batch_vector(self)
+            query._read_global_index_result(result)
+            for result in query._execute_batch_vector(query)
         ]
 
     def to_pandas(self):
@@ -614,14 +668,17 @@ def to_list(self) -> List[List[dict]]:
         return [table.to_pylist() for table in self.to_arrow()]
 
     def _execute_batch_vector(self, query):
-        limit = query._limit if query._limit is not None else 10
+        return query._search_builder().execute_batch_local()
+
+    def _search_builder(self):
+        limit = self._limit if self._limit is not None else 10
         builder = (
             self._table.new_batch_vector_search_builder()
             .with_vector_column(self._vector_column)
             .with_query_vectors(self._vectors)
             .with_limit(limit)
             .with_options(self._vector_options)
         )
-        if query._pre_filter is not None:
-            builder = builder.with_filter(query._pre_filter)
-        return builder.execute_batch_local()
+        if self._pre_filter is not None:
+            builder = builder.with_filter(self._pre_filter)
+        return builder

paimon-python/pypaimon/read/table_scan.py

@@ -447,9 +447,7 @@ def incremental_manifest():
 
         if has_time_travel:
             def time_travel_manifest_scanner():
-                snapshot = TimeTravelUtil.try_travel_to_snapshot(
-                    options, self.table.tag_manager(), snapshot_manager
-                )
+                snapshot = TimeTravelUtil.resolve_snapshot(self.table)
                 if snapshot is None:
                     raise ValueError(
                         "Could not resolve time travel snapshot from scan options."
@@ -466,7 +464,7 @@ def time_travel_manifest_scanner():
             )
 
         def all_manifests():
-            snapshot = snapshot_manager.get_latest_snapshot()
+            snapshot = TimeTravelUtil.resolve_snapshot(self.table)
             return manifest_list_manager.read_all(snapshot), snapshot
 
         return FileScanner(

paimon-python/pypaimon/snapshot/time_travel_util.py

@@ -66,6 +66,16 @@ def _parse_timestamp_to_millis(timestamp_str: str) -> int:
 class TimeTravelUtil:
     """The util class of resolve snapshot from scan params for time travel."""
 
+    @staticmethod
+    def resolve_snapshot(table):
+        """Resolve the table's read view, including an explicitly pinned empty view."""
+        if hasattr(table, "_read_snapshot"):
+            return table._read_snapshot
+        manager = table.snapshot_manager()
+        snapshot = TimeTravelUtil.try_travel_to_snapshot(
+            table.options.options, table.tag_manager(), manager)
+        return snapshot if snapshot is not None else manager.get_latest_snapshot()
+
     @staticmethod
     def try_travel_to_snapshot(
             options: Options,

paimon-python/pypaimon/table/file_store_table.py

@@ -516,6 +516,22 @@ def copy_without_time_travel(self, options: dict) -> 'FileStoreTable':
         """Copy this table while preserving its already resolved schema."""
         return self._copy(options, resolve_time_travel=False)
 
+    def _copy_with_snapshot(self, snapshot):
+        """Keep one resolved read view, including tag metadata and empty tables."""
+        from pypaimon.snapshot.time_travel_util import SCAN_KEYS
+        options = {key: None for key in SCAN_KEYS if key in self.table_schema.options}
+        options[CoreOptions.SCAN_MODE.key()] = "from-snapshot" if snapshot is not None else "default"
+        if snapshot is not None:
+            options[CoreOptions.SCAN_SNAPSHOT_ID.key()] = str(snapshot.id)
+        # Native planning cannot consume retained tag metadata or a pinned empty view.
+        # scan.version can resolve to a tag as well.
+        if snapshot is None or any(option.key() in self.table_schema.options for option in (
+                CoreOptions.SCAN_TAG_NAME, CoreOptions.SCAN_VERSION)):
+            options[CoreOptions.SCAN_NATIVE_PLAN_ENABLED.key()] = "false"
+        table = self.copy_without_time_travel(options)
+        table._read_snapshot = snapshot
+        return table
+
     def _copy(self, options: dict, resolve_time_travel: bool) -> 'FileStoreTable':
         if CoreOptions.BUCKET.key() in options and int(options.get(CoreOptions.BUCKET.key())) != self.options.bucket():
             raise ValueError("Cannot change bucket number")
@@ -531,7 +547,12 @@ def _copy(self, options: dict, resolve_time_travel: bool) -> 'FileStoreTable':
         # Cumulative copy() overrides (removals kept as None) vs the on-disk schema.
         applied_options = {**getattr(self, '_applied_dynamic_options', {}), **options}
 
-        if resolve_time_travel:
+        from pypaimon.snapshot.time_travel_util import SCAN_KEYS
+        preserve_snapshot = hasattr(self, "_read_snapshot") and not any(
+            key in options for key in SCAN_KEYS + [
+                CoreOptions.SCAN_MODE.key(), CoreOptions.BRANCH.key(),
+                CoreOptions.INCREMENTAL_BETWEEN_TIMESTAMP.key()])
+        if resolve_time_travel and not preserve_snapshot:
             time_travel_schema = self._try_time_travel(Options(new_options), set(applied_options))
             if time_travel_schema is not None:
                 new_table_schema = time_travel_schema
@@ -554,6 +575,8 @@ def _copy(self, options: dict, resolve_time_travel: bool) -> 'FileStoreTable':
         new_table = FileStoreTable(self.file_io, new_identifier, self.table_path,
                                    new_table_schema, catalog_env)
         new_table._applied_dynamic_options = applied_options
+        if preserve_snapshot:
+            new_table._read_snapshot = self._read_snapshot
         return new_table
 
     def _try_time_travel(self, options: Options, dynamic_option_keys: Set[str]) -> Optional[TableSchema]:

paimon-python/pypaimon/table/source/batch_vector_search_builder.py

@@ -18,6 +18,7 @@
 """Builder to build batch vector search over multiple query vectors."""
 
 from abc import ABC, abstractmethod
+from pypaimon.table.source.search_diagnostics import SearchDiagnostics
 
 from pypaimon.table.source.vector_search_builder import (
     AbstractVectorSearchBuilderImpl,
@@ -28,9 +29,11 @@
 from pypaimon.table.source.vector_search_read import BatchVectorSearchRead  # noqa: F401
 
 
-class BatchVectorSearchBuilder(ABC):
+class BatchVectorSearchBuilder(SearchDiagnostics, ABC):
     """Builder to build batch vector search; result ``i`` matches vector ``i``."""
 
+    _diagnostic_kind = "batch_vector"
+
     @abstractmethod
     def with_limit(self, limit):
         # type: (int) -> BatchVectorSearchBuilder