This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 67d88bf5 [Core] Add snapshot-pinned full-text search with scalar
filters (#1095)
67d88bf5 is described below
commit 67d88bf5c78b25d23b5cbe4099a1aa3857d24ffd
Author: Jingsong Lee <[email protected]>
AuthorDate: Sat Oct 10 20:59:36 2026 +0800
[Core] Add snapshot-pinned full-text search with scalar filters (#1095)
---
.../python/python/pypaimon_rust/datafusion.pyi | 23 ++
bindings/python/src/context.rs | 5 +
bindings/python/src/full_text_search.rs | 192 +++++++++
bindings/python/src/lib.rs | 1 +
bindings/python/src/table.rs | 4 +
crates/paimon/src/table/de_vector_read.rs | 3 +-
crates/paimon/src/table/de_vector_scan.rs | 66 +--
crates/paimon/src/table/de_vector_scan/tests.rs | 1 +
.../paimon/src/table/full_text_search_builder.rs | 239 ++++++-----
.../src/table/full_text_search_builder/filter.rs | 206 ++++++++++
.../table/full_text_search_builder/parity_tests.rs | 449 +++++++++++++++++++++
.../src/table/full_text_search_builder/read.rs | 89 ++++
.../src/table/full_text_search_builder/scan.rs | 183 +++++++++
.../paimon/src/table/global_index_search_filter.rs | 77 ++++
crates/paimon/src/table/mod.rs | 5 +-
docs/src/sql.md | 34 ++
16 files changed, 1416 insertions(+), 161 deletions(-)
diff --git a/bindings/python/python/pypaimon_rust/datafusion.pyi
b/bindings/python/python/pypaimon_rust/datafusion.pyi
index e38d5e5e..9279e894 100644
--- a/bindings/python/python/pypaimon_rust/datafusion.pyi
+++ b/bindings/python/python/pypaimon_rust/datafusion.pyi
@@ -221,6 +221,7 @@ class Table:
def schema(self) -> TableSchema: ...
def new_read_builder(self, options: Optional[Dict[str, str]] = None) ->
ReadBuilder: ...
def new_stream_scan(self) -> StreamTableScan: ...
+ def new_full_text_search_builder(self) -> FullTextSearchBuilder: ...
def new_vector_search_builder(self) -> "VectorSearchBuilder": ...
def new_batch_vector_search_builder(self) -> "BatchVectorSearchBuilder":
...
def new_global_index_build_builder(self) -> "GlobalIndexBuildBuilder": ...
@@ -613,3 +614,25 @@ class SearchResultReadBuilder:
def read(self) -> List[pyarrow.RecordBatch]:
"""Read user columns and scores in relevance order from the search
snapshot."""
...
+
+class FullTextSearchBuilder:
+ def with_query(self, field_name: str, query: str) ->
FullTextSearchBuilder: ...
+ def with_limit(self, limit: int) -> FullTextSearchBuilder: ...
+ def with_filter(self, predicate: Dict[str, Any]) -> FullTextSearchBuilder:
...
+ def with_partition_filter(self, predicate: Dict[str, Any]) ->
FullTextSearchBuilder: ...
+ def new_full_text_scan(self) -> FullTextScan: ...
+ def new_full_text_read(self) -> FullTextRead: ...
+ def execute_local(self) -> FullTextSearchResult: ...
+
+class FullTextScan:
+ def scan(self) -> FullTextScanPlan: ...
+
+class FullTextScanPlan:
+ def snapshot_id(self) -> Optional[int]: ...
+
+class FullTextRead:
+ def read_plan(self, plan: FullTextScanPlan) -> FullTextSearchResult: ...
+
+class FullTextSearchResult:
+ def __len__(self) -> int: ...
+ def row_ids(self) -> Dict[int, float]: ...
diff --git a/bindings/python/src/context.rs b/bindings/python/src/context.rs
index f0baad1b..4c6637a4 100644
--- a/bindings/python/src/context.rs
+++ b/bindings/python/src/context.rs
@@ -727,6 +727,11 @@ pub fn register_module(py: Python<'_>, m: &Bound<'_,
PyModule>) -> PyResult<()>
this.add_class::<PaimonCatalog>()?;
this.add_class::<crate::table::PyTable>()?;
this.add_class::<crate::index_build::PyGlobalIndexBuildBuilder>()?;
+ this.add_class::<crate::full_text_search::PyFullTextSearchBuilder>()?;
+ this.add_class::<crate::full_text_search::PyFullTextScan>()?;
+ this.add_class::<crate::full_text_search::PyFullTextScanPlan>()?;
+ this.add_class::<crate::full_text_search::PyFullTextRead>()?;
+ this.add_class::<crate::full_text_search::PyFullTextSearchResult>()?;
this.add_class::<crate::vector_search::PyVectorSearchBuilder>()?;
this.add_class::<crate::vector_search::PyBatchVectorSearchBuilder>()?;
this.add_class::<crate::vector_search::PyVectorScan>()?;
diff --git a/bindings/python/src/full_text_search.rs
b/bindings/python/src/full_text_search.rs
new file mode 100644
index 00000000..b73c0fb6
--- /dev/null
+++ b/bindings/python/src/full_text_search.rs
@@ -0,0 +1,192 @@
+// 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.
+
+//! Thin bindings over core full-text Scan -> Plan -> Read.
+
+use crate::error::to_py_err;
+use crate::predicate::dict_to_table_predicate;
+use paimon::spec::Predicate;
+use paimon::table::{FullTextRead, FullTextScan, FullTextScanPlan, Table};
+use paimon_datafusion::runtime::runtime;
+use pyo3::prelude::*;
+use pyo3::types::PyDict;
+use std::collections::HashMap;
+use std::sync::Arc;
+
+#[derive(Default)]
+struct Config {
+ query: Option<(String, String)>,
+ limit: Option<usize>,
+ filters: Vec<Predicate>,
+}
+
+impl Config {
+ fn builder<'a>(
+ &self,
+ table: &'a Table,
+ ) -> paimon::Result<paimon::table::FullTextSearchBuilder<'a>> {
+ let mut builder = table.new_full_text_search_builder();
+ if let Some((column, query)) = &self.query {
+ builder.with_query(column, query);
+ }
+ if let Some(limit) = self.limit {
+ builder.with_limit(limit);
+ }
+ for filter in &self.filters {
+ builder.with_filter(filter.clone());
+ }
+ Ok(builder)
+ }
+}
+
+#[pyclass(name = "FullTextSearchBuilder", module = "pypaimon_rust.datafusion")]
+pub struct PyFullTextSearchBuilder {
+ table: Arc<Table>,
+ config: Config,
+}
+
+impl PyFullTextSearchBuilder {
+ pub(crate) fn new(table: Arc<Table>) -> Self {
+ Self {
+ table,
+ config: Config::default(),
+ }
+ }
+}
+
+#[pymethods]
+impl PyFullTextSearchBuilder {
+ fn with_query(
+ mut slf: PyRefMut<'_, Self>,
+ field_name: String,
+ query: String,
+ ) -> PyRefMut<'_, Self> {
+ slf.config.query = Some((field_name, query));
+ slf
+ }
+ fn with_limit(mut slf: PyRefMut<'_, Self>, limit: usize) -> PyRefMut<'_,
Self> {
+ slf.config.limit = Some(limit);
+ slf
+ }
+ fn with_filter<'py>(
+ mut slf: PyRefMut<'py, Self>,
+ predicate: &Bound<'_, PyDict>,
+ ) -> PyResult<PyRefMut<'py, Self>> {
+ let filter = dict_to_table_predicate(predicate, slf.table.schema(),
true)?;
+ slf.config.filters.push(filter);
+ Ok(slf)
+ }
+ fn with_partition_filter<'py>(
+ mut slf: PyRefMut<'py, Self>,
+ predicate: &Bound<'_, PyDict>,
+ ) -> PyResult<PyRefMut<'py, Self>> {
+ let filter = dict_to_table_predicate(predicate, slf.table.schema(),
true)?;
+ slf.config
+ .builder(&slf.table)
+ .map_err(to_py_err)?
+ .with_partition_filter(filter.clone())
+ .map_err(to_py_err)?;
+ slf.config.filters.push(filter);
+ Ok(slf)
+ }
+ fn new_full_text_scan(&self) -> PyResult<PyFullTextScan> {
+ Ok(PyFullTextScan {
+ inner: self
+ .config
+ .builder(&self.table)
+ .map_err(to_py_err)?
+ .new_scan()
+ .map_err(to_py_err)?,
+ })
+ }
+ fn new_full_text_read(&self) -> PyResult<PyFullTextRead> {
+ Ok(PyFullTextRead {
+ inner: self
+ .config
+ .builder(&self.table)
+ .map_err(to_py_err)?
+ .new_read()
+ .map_err(to_py_err)?,
+ })
+ }
+ fn execute_local(&self, py: Python<'_>) ->
PyResult<PyFullTextSearchResult> {
+ let builder = self.config.builder(&self.table).map_err(to_py_err)?;
+ py.detach(|| runtime().block_on(builder.execute_scored()))
+ .map(|inner| PyFullTextSearchResult { inner })
+ .map_err(to_py_err)
+ }
+}
+
+#[pyclass(name = "FullTextScan", module = "pypaimon_rust.datafusion")]
+pub struct PyFullTextScan {
+ inner: FullTextScan,
+}
+#[pymethods]
+impl PyFullTextScan {
+ fn scan(&self, py: Python<'_>) -> PyResult<PyFullTextScanPlan> {
+ py.detach(|| runtime().block_on(self.inner.scan()))
+ .map(|inner| PyFullTextScanPlan { inner })
+ .map_err(to_py_err)
+ }
+}
+
+#[pyclass(name = "FullTextScanPlan", module = "pypaimon_rust.datafusion")]
+pub struct PyFullTextScanPlan {
+ inner: FullTextScanPlan,
+}
+#[pymethods]
+impl PyFullTextScanPlan {
+ fn snapshot_id(&self) -> Option<i64> {
+ self.inner.snapshot_id()
+ }
+}
+
+#[pyclass(name = "FullTextRead", module = "pypaimon_rust.datafusion")]
+pub struct PyFullTextRead {
+ inner: FullTextRead,
+}
+#[pymethods]
+impl PyFullTextRead {
+ fn read_plan(
+ &self,
+ py: Python<'_>,
+ plan: &PyFullTextScanPlan,
+ ) -> PyResult<PyFullTextSearchResult> {
+ py.detach(|| runtime().block_on(self.inner.read(plan.inner.clone())))
+ .map(|inner| PyFullTextSearchResult { inner })
+ .map_err(to_py_err)
+ }
+}
+
+#[pyclass(name = "FullTextSearchResult", module = "pypaimon_rust.datafusion")]
+pub struct PyFullTextSearchResult {
+ inner: paimon::full_text::SearchResult,
+}
+#[pymethods]
+impl PyFullTextSearchResult {
+ fn __len__(&self) -> usize {
+ self.inner.len()
+ }
+ fn row_ids(&self) -> HashMap<u64, f32> {
+ self.inner
+ .row_ids
+ .iter()
+ .copied()
+ .zip(self.inner.scores.iter().copied())
+ .collect()
+ }
+}
diff --git a/bindings/python/src/lib.rs b/bindings/python/src/lib.rs
index 2478c02d..e4a5c027 100644
--- a/bindings/python/src/lib.rs
+++ b/bindings/python/src/lib.rs
@@ -21,6 +21,7 @@ mod blob;
mod blob_uri_reader;
mod context;
mod error;
+mod full_text_search;
mod index_build;
mod merge_into;
mod oss_cpp;
diff --git a/bindings/python/src/table.rs b/bindings/python/src/table.rs
index 363c4a07..e0daf6ea 100644
--- a/bindings/python/src/table.rs
+++ b/bindings/python/src/table.rs
@@ -198,6 +198,10 @@ impl PyTable {
}
}
+ fn new_full_text_search_builder(&self) ->
crate::full_text_search::PyFullTextSearchBuilder {
+
crate::full_text_search::PyFullTextSearchBuilder::new(Arc::clone(&self.inner))
+ }
+
fn new_vector_search_builder(&self) ->
crate::vector_search::PyVectorSearchBuilder {
crate::vector_search::PyVectorSearchBuilder::new(Arc::clone(&self.inner))
}
diff --git a/crates/paimon/src/table/de_vector_read.rs
b/crates/paimon/src/table/de_vector_read.rs
index 665953bc..ae72aa1c 100644
--- a/crates/paimon/src/table/de_vector_read.rs
+++ b/crates/paimon/src/table/de_vector_read.rs
@@ -24,11 +24,12 @@ use crate::spec::{
row_id_data_field, CoreOptions, DataField, DataType, FileKind,
GlobalIndexSearchMode,
IndexFileMeta, IndexManifestEntry, Predicate, ROW_ID_FIELD_NAME,
};
-use crate::table::de_vector_scan::{matching_row_ids_for_filter,
DeVectorScanPlan};
+use crate::table::de_vector_scan::DeVectorScanPlan;
use crate::table::global_index_scanner::{
deleted_row_ranges_for_data_evolution_dvs, search_limit_with_deleted_rows,
unindexed_ranges_for_global_index_entries, GlobalIndexScanner,
RowRangeIndex,
};
+use crate::table::global_index_search_filter::matching_row_ids_for_filter;
use crate::table::global_index_types::normalize_queryable_global_index_type;
use crate::table::index_file_path::IndexFileLocation;
use crate::table::pk_vector_position_read::SEARCH_SCORE_COLUMN;
diff --git a/crates/paimon/src/table/de_vector_scan.rs
b/crates/paimon/src/table/de_vector_scan.rs
index ac154ef2..883dd077 100644
--- a/crates/paimon/src/table/de_vector_scan.rs
+++ b/crates/paimon/src/table/de_vector_scan.rs
@@ -17,14 +17,13 @@
//! Plans global-index vector searches against one snapshot and resolves
scalar pre-filters.
-use crate::spec::{CoreOptions, IndexManifest, IndexManifestEntry, Predicate,
ROW_ID_FIELD_NAME};
+use crate::spec::{CoreOptions, IndexManifest, IndexManifestEntry, Predicate};
use crate::table::bucket_filter::split_partition_and_data_predicates;
+use crate::table::global_index_search_filter::matching_row_ids_for_filter;
use crate::table::partition_filter::PartitionFilter;
use crate::table::vector_scan::Scan;
-use crate::table::{RowRange, Table};
+use crate::table::Table;
use crate::vindex::vector_search_timing_enabled;
-use arrow_array::{Array, Int64Array};
-use futures::TryStreamExt;
use roaring::RoaringTreemap;
use std::sync::Arc;
use std::time::{Duration, Instant};
@@ -54,65 +53,6 @@ fn same_vector_search_table(left: &Table, right: &Table) ->
bool {
&& left.branch() == right.branch()
}
-pub(super) async fn matching_row_ids_for_filter(
- table: &Table,
- filter: &Predicate,
- ranges: Option<Vec<RowRange>>,
-) -> crate::Result<RoaringTreemap> {
- let mut read_builder = table.new_read_builder();
- read_builder
- .with_projection(&[ROW_ID_FIELD_NAME])?
- .with_filter(filter.clone());
- if let Some(ranges) = ranges {
- read_builder.with_row_ranges(ranges);
- }
- let plan = read_builder.new_scan().plan().await?;
- let read = read_builder.new_read()?;
- let mut stream = read.to_arrow(plan.splits())?;
- let mut row_ids = RoaringTreemap::new();
- while let Some(batch) = stream.try_next().await? {
- let index =
- batch
- .schema()
- .index_of(ROW_ID_FIELD_NAME)
- .map_err(|_| crate::Error::DataInvalid {
- message: format!(
- "scalar vector pre-filter read is missing
{ROW_ID_FIELD_NAME}"
- ),
- source: None,
- })?;
- let values = batch
- .column(index)
- .as_any()
- .downcast_ref::<Int64Array>()
- .ok_or_else(|| crate::Error::DataInvalid {
- message: format!(
- "scalar vector pre-filter {ROW_ID_FIELD_NAME} column is
not Int64"
- ),
- source: None,
- })?;
- for row in 0..values.len() {
- if values.is_null(row) {
- return Err(crate::Error::DataInvalid {
- message: format!(
- "scalar vector pre-filter produced a null
{ROW_ID_FIELD_NAME}"
- ),
- source: None,
- });
- }
- let row_id = values.value(row);
- let row_id = u64::try_from(row_id).map_err(|_|
crate::Error::DataInvalid {
- message: format!(
- "scalar vector pre-filter produced a negative
{ROW_ID_FIELD_NAME}: {row_id}"
- ),
- source: None,
- })?;
- row_ids.insert(row_id);
- }
- }
- Ok(row_ids)
-}
-
impl Table {
/// Resolve a scalar predicate once and pin all later vector-search/read
/// stages to the same snapshot.
diff --git a/crates/paimon/src/table/de_vector_scan/tests.rs
b/crates/paimon/src/table/de_vector_scan/tests.rs
index 310a442f..5033ed9b 100644
--- a/crates/paimon/src/table/de_vector_scan/tests.rs
+++ b/crates/paimon/src/table/de_vector_scan/tests.rs
@@ -27,6 +27,7 @@ use crate::vector_search::SearchResult;
use arrow_array::builder::{Float32Builder, ListBuilder};
use arrow_array::{ArrayRef, Float32Array, Int32Array, RecordBatch};
use arrow_schema::{DataType as ArrowDataType, Field as ArrowField, Schema as
ArrowSchema};
+use futures::TryStreamExt;
use roaring::RoaringTreemap;
use std::collections::HashMap;
use std::sync::Arc;
diff --git a/crates/paimon/src/table/full_text_search_builder.rs
b/crates/paimon/src/table/full_text_search_builder.rs
index 1c567162..8255768c 100644
--- a/crates/paimon/src/table/full_text_search_builder.rs
+++ b/crates/paimon/src/table/full_text_search_builder.rs
@@ -23,13 +23,13 @@ use crate::full_text::{FullTextSearch, SearchResult};
use crate::io::FileIO;
use crate::spec::{
CoreOptions, DataField, FileKind, GlobalIndexSearchMode, IndexFileMeta,
IndexManifest,
- IndexManifestEntry, ROW_ID_FIELD_NAME,
+ IndexManifestEntry, Predicate, ROW_ID_FIELD_NAME,
};
use crate::table::data_file_reader::DataFileReader;
use crate::table::full_text_index_adapter::{search_full_text_file,
search_full_text_index};
use crate::table::global_index_scanner::{
- deleted_row_ranges_for_data_evolution_dvs, search_limit_with_deleted_rows,
- unindexed_ranges_for_global_index_entries, RowRangeIndex,
+ deleted_row_ranges_for_data_evolution_dvs,
unindexed_ranges_for_global_index_entries,
+ RowRangeIndex,
};
use crate::table::index_file_path::IndexFileLocation;
use crate::table::pk_full_text_read::PrimaryKeyFullTextRead;
@@ -46,7 +46,11 @@ use serde_json::json;
use std::collections::{HashMap, HashSet};
const FULL_TEXT_INDEX_TYPE: &str = "full-text";
-const FULL_TEXT_INDEX_SEARCH_CONCURRENCY: usize = 8;
+mod filter;
+mod read;
+mod scan;
+pub use read::FullTextRead;
+pub use scan::{FullTextScan, FullTextScanPlan};
/// Builder for executing full-text search on a Paimon table.
///
@@ -65,8 +69,10 @@ pub struct FullTextSearchBuilder<'a> {
table: &'a Table,
text_column: Option<String>,
query_text: Option<String>,
+ query_is_dsl: bool,
limit: Option<usize>,
include_row_ids: Option<RoaringTreemap>,
+ filter: Option<Predicate>,
}
impl<'a> FullTextSearchBuilder<'a> {
@@ -75,8 +81,10 @@ impl<'a> FullTextSearchBuilder<'a> {
table,
text_column: None,
query_text: None,
+ query_is_dsl: false,
limit: None,
include_row_ids: None,
+ filter: None,
}
}
@@ -89,6 +97,7 @@ impl<'a> FullTextSearchBuilder<'a> {
/// Set the query text to search for.
pub fn with_query_text(&mut self, query: &str) -> &mut Self {
self.query_text = Some(query.to_string());
+ self.query_is_dsl = false;
self
}
@@ -115,77 +124,35 @@ impl<'a> FullTextSearchBuilder<'a> {
}
pub async fn execute_scored(&self) -> crate::Result<SearchResult> {
- // Fail closed: returns data-derived row ranges outside
`TableScan`/`TableRead`.
- let core = CoreOptions::new(self.table.schema().options());
- core.ensure_read_authorized()?;
- let text_column =
- self.text_column
- .as_deref()
- .ok_or_else(|| crate::Error::ConfigInvalid {
- message: "Text column must be set via
with_text_column()".to_string(),
- })?;
- let query_text = self
- .query_text
- .as_deref()
- .ok_or_else(|| crate::Error::ConfigInvalid {
- message: "Query text must be set via
with_query_text()".to_string(),
- })?;
- let limit = self.limit.ok_or_else(|| crate::Error::ConfigInvalid {
- message: "Limit must be set via with_limit()".to_string(),
- })?;
-
- // Primary-key full-text search does not produce global row-ids: it
maps
- // hits to physical `(data file, row position)` pairs. A
scored/row-range
- // search is therefore unsupported on the PK path — callers must use
- // `execute_read`. Fail loud rather than fall through to the append/DE
- // global-index path (which would search the wrong index and could
return
- // an empty or wrong result). Mirrors the vector builder.
- if resolves_to_pk_full_text_path(&core, text_column) {
- return Err(crate::Error::DataInvalid {
- message: "primary-key full-text search does not produce global
row ids; use the \
- materialized read (execute_read) instead"
- .to_string(),
- source: None,
- });
- }
+ let read = self.new_read()?;
+ read.read(self.new_scan()?.scan().await?).await
+ }
- let mut search = FullTextSearch::new(
- normalize_query_text(query_text, text_column)?,
- limit,
- text_column.to_string(),
- )?;
- if let Some(include_row_ids) = &self.include_row_ids {
- search = search.with_include_row_ids(include_row_ids.clone());
- }
+ /// Configure the queried field and query, matching Java withQuery.
+ pub fn with_query(&mut self, field_name: &str, query: &str) -> &mut Self {
+ self.with_text_column(field_name);
+ self.query_text = Some(query.to_string());
+ self.query_is_dsl = true;
+ self
+ }
- let snapshot_manager = self.table.snapshot_manager();
+ /// Exclude rows before Top-K; partition conjuncts prune the manifest.
+ pub fn with_filter(&mut self, filter: Predicate) -> &mut Self {
+ super::vector_search_common::add_filter(&mut self.filter, filter);
+ self
+ }
- let snapshot = match snapshot_manager.get_latest_snapshot().await? {
- Some(s) => s,
- None => return Ok(SearchResult::empty()),
- };
+ pub fn with_partition_filter(&mut self, filter: Predicate) ->
crate::Result<&mut Self> {
+ super::partition_filter::validate_partition_filter(self.table,
&filter)?;
+ Ok(self.with_filter(filter))
+ }
- let index_entries = match snapshot.index_manifest() {
- Some(index_manifest_name) => {
- let manifest_path =
snapshot_manager.manifest_path(index_manifest_name);
- IndexManifest::read(self.table.file_io(),
&manifest_path).await?
- }
- None => Vec::new(),
- };
+ pub fn new_scan(&self) -> crate::Result<FullTextScan> {
+ FullTextScan::new(self)
+ }
- evaluate_full_text_search(
- FullTextSearchEvaluation {
- table: Some(self.table),
- file_io: self.table.file_io(),
- table_path: self.table.location(),
- table_options: self.table.schema().options(),
- schema_fields: self.table.schema().fields(),
- next_row_id: snapshot.next_row_id(),
- },
- &index_entries,
- &search,
- )
- .await
+ pub fn new_read(&self) -> crate::Result<FullTextRead> {
+ FullTextRead::new(self)
}
/// Run the full-text search and materialize the matching rows as Arrow
batches,
@@ -256,7 +223,20 @@ impl<'a> FullTextSearchBuilder<'a> {
source: None,
})?;
- let plan = PrimaryKeyFullTextScan::new(self.table, field_id, None)
+ if let Some(filter) = &self.filter {
+ let (_, data) =
super::bucket_filter::split_partition_and_data_predicates(
+ filter.clone(),
+ self.table.schema().fields(),
+ self.table.schema().partition_keys(),
+ );
+ if !data.is_empty() {
+ return Err(crate::Error::Unsupported {
+ message: "Primary-key full-text search does not support
non-partition filters"
+ .into(),
+ });
+ }
+ }
+ let plan = PrimaryKeyFullTextScan::new(self.table, field_id,
self.filter.clone())
.plan()
.await?;
if plan.splits.is_empty() {
@@ -309,6 +289,8 @@ struct FullTextSearchEvaluation<'a> {
table_options: &'a HashMap<String, String>,
schema_fields: &'a [DataField],
next_row_id: Option<i64>,
+ partition_filter: Option<&'a Predicate>,
+ row_filter: Option<&'a Predicate>,
}
async fn evaluate_full_text_search(
@@ -339,7 +321,7 @@ async fn evaluate_full_text_search(
.collect();
let has_fulltext_entries = !fulltext_entries.is_empty();
- if !has_fulltext_entries && search_mode == GlobalIndexSearchMode::Fast {
+ if !has_fulltext_entries {
return Ok(SearchResult::empty());
}
@@ -356,11 +338,24 @@ async fn evaluate_full_text_search(
None
};
+ let mut indexed_search = search.clone();
+ if let Some(filter) = evaluation.row_filter {
+ let matched =
+ filter::matching_indexed_rows(&evaluation, index_entries,
&fulltext_entries, filter)
+ .await?;
+ indexed_search.include_row_ids = Some(filter::intersect_include(
+ search.include_row_ids.as_ref(),
+ matched,
+ ));
+ }
let mut merged = SearchResult::empty();
if !fulltext_entries.is_empty() {
- let plans =
- plan_full_text_index_searches(fulltext_entries,
search.include_row_ids.as_ref())?;
- let results = futures::stream::iter(plans)
+ let plans = plan_full_text_index_searches(
+ fulltext_entries,
+ indexed_search.include_row_ids.as_ref(),
+ )?;
+ let results = plans
+ .into_iter()
.map(|plan| {
let entry = plan.entry;
let global_meta =
entry.index_file.global_index_meta.as_ref().unwrap();
@@ -373,22 +368,22 @@ async fn evaluate_full_text_search(
let local_filter = plan.local_filter;
let row_range_start = global_meta.row_range_start;
let row_range_end = global_meta.row_range_end;
- let limit = search_limit_with_deleted_rows(
- search.limit,
- row_range_start,
- row_range_end,
- deleted_row_index.as_ref(),
- );
- async move {
+ // Compound query leaves must see every candidate in the shard
+ // before Boolean/boost composition and the final global Top-K.
+ let limit = candidate_limit(row_range_start, row_range_end)?;
+ Ok::<_, crate::Error>(async move {
let input = evaluation.file_io.new_input(&path)?;
let reader = Box::new(input.reader().await?);
let result =
search_full_text_index(reader, file_name, query_text,
limit, local_filter)
.await?;
Ok::<_,
crate::Error>(search_result_from_core(result)?.offset(row_range_start))
- }
+ })
})
- .buffered(FULL_TEXT_INDEX_SEARCH_CONCURRENCY)
+ .collect::<crate::Result<Vec<_>>>()?
+ .into_iter();
+ let results = futures::stream::iter(results)
+ .buffered(core_options.global_index_thread_num()?)
.try_collect::<Vec<_>>()
.await?;
for r in &results {
@@ -402,7 +397,7 @@ async fn evaluate_full_text_search(
message: "Full-text raw search in detail mode requires table
context".to_string(),
source: None,
})?;
- detail_data_ranges_for_table(table).await?
+ detail_data_ranges_for_table(table,
evaluation.partition_filter).await?
} else {
Vec::new()
};
@@ -420,9 +415,23 @@ async fn evaluate_full_text_search(
message: "Full-text raw search requires table
context".to_string(),
source: None,
})?;
- let raw_result =
- read_raw_full_text_search(table, search, &raw_ranges,
evaluation.table_options)
- .await?;
+ let mut raw_search = search.clone();
+ if let Some(filter) = evaluation.row_filter {
+ let matched =
+ filter::matching_rows(&evaluation, filter,
raw_ranges.clone()).await?;
+ raw_search.include_row_ids = Some(filter::intersect_include(
+ search.include_row_ids.as_ref(),
+ matched,
+ ));
+ }
+ let raw_result = read_raw_full_text_search_filtered(
+ table,
+ &raw_search,
+ &raw_ranges,
+ evaluation.table_options,
+ evaluation.partition_filter,
+ )
+ .await?;
merged = merged.without_row_ranges(&raw_ranges)?;
merged = merged.or(&raw_result);
}
@@ -517,13 +526,15 @@ fn is_full_text_index_file(index_file: &IndexFileMeta) ->
bool {
index_file.index_type == FULL_TEXT_INDEX_TYPE
}
-async fn detail_data_ranges_for_table(table: &Table) ->
crate::Result<Vec<RowRange>> {
- let plan = table
- .new_read_builder()
- .new_scan()
- .with_scan_all_files()
- .plan()
- .await?;
+async fn detail_data_ranges_for_table(
+ table: &Table,
+ partition_filter: Option<&Predicate>,
+) -> crate::Result<Vec<RowRange>> {
+ let mut builder = table.new_read_builder();
+ if let Some(filter) = partition_filter {
+ builder.with_filter(filter.clone());
+ }
+ let plan = builder.new_scan().with_scan_all_files().plan().await?;
let mut ranges = Vec::new();
for split in plan.splits() {
for file in split.data_files() {
@@ -535,11 +546,22 @@ async fn detail_data_ranges_for_table(table: &Table) ->
crate::Result<Vec<RowRan
Ok(merge_row_ranges(ranges))
}
+#[cfg(test)]
async fn read_raw_full_text_search(
table: &Table,
search: &FullTextSearch,
raw_ranges: &[RowRange],
table_options: &HashMap<String, String>,
+) -> crate::Result<SearchResult> {
+ read_raw_full_text_search_filtered(table, search, raw_ranges,
table_options, None).await
+}
+
+async fn read_raw_full_text_search_filtered(
+ table: &Table,
+ search: &FullTextSearch,
+ raw_ranges: &[RowRange],
+ table_options: &HashMap<String, String>,
+ partition_filter: Option<&Predicate>,
) -> crate::Result<SearchResult> {
if raw_ranges.is_empty() {
return Ok(SearchResult::empty());
@@ -555,6 +577,9 @@ async fn read_raw_full_text_search(
read_builder
.with_projection(&[search.field_name.as_str(), ROW_ID_FIELD_NAME])?
.with_row_ranges(raw_ranges);
+ if let Some(filter) = partition_filter {
+ read_builder.with_filter(filter.clone());
+ }
let plan = read_builder.new_scan().plan().await?;
if plan.splits().is_empty() {
return Ok(SearchResult::empty());
@@ -622,7 +647,7 @@ async fn read_raw_full_text_search(
index_file,
"raw-full-text.index".to_string(),
search.query_text.clone(),
- search.limit,
+ candidate_limit(row_range_start, row_range_end)?,
local_filter,
)
.await?;
@@ -636,6 +661,19 @@ fn raw_index_task_error(error: tokio::task::JoinError) ->
crate::Error {
}
}
+fn candidate_limit(from: i64, to: i64) -> crate::Result<usize> {
+ if from < 0 || to < from {
+ return Err(crate::Error::DataInvalid {
+ message: format!("Invalid full-text row range [{from}, {to}]"),
+ source: None,
+ });
+ }
+ usize::try_from(to as u64 - from as u64 + 1).map_err(|_|
crate::Error::DataInvalid {
+ message: "Full-text row range exceeds the native candidate
limit".into(),
+ source: None,
+ })
+}
+
fn raw_full_text_batch_documents(
batch: &RecordBatch,
search: &FullTextSearch,
@@ -889,6 +927,8 @@ mod tests {
table_options: &options,
schema_fields: &fields,
next_row_id: Some(10),
+ partition_filter: None,
+ row_filter: None,
},
&[],
&search,
@@ -1042,6 +1082,8 @@ mod tests {
table_options: &HashMap::new(),
schema_fields: &fields,
next_row_id: Some(102),
+ partition_filter: None,
+ row_filter: None,
},
&[entry],
&search,
@@ -1126,6 +1168,8 @@ mod tests {
table_options: &HashMap::new(),
schema_fields: &fields,
next_row_id: Some(102),
+ partition_filter: None,
+ row_filter: None,
},
&[entry, missing_empty_filter_entry],
&search,
@@ -1519,3 +1563,6 @@ mod tests {
);
}
}
+
+#[cfg(test)]
+mod parity_tests;
diff --git a/crates/paimon/src/table/full_text_search_builder/filter.rs
b/crates/paimon/src/table/full_text_search_builder/filter.rs
new file mode 100644
index 00000000..37246cca
--- /dev/null
+++ b/crates/paimon/src/table/full_text_search_builder/filter.rs
@@ -0,0 +1,206 @@
+// 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.
+
+//! Java full-text scalar coverage and exact candidate refinement.
+
+use super::*;
+use crate::table::global_index_scanner::GlobalIndexScanner;
+use crate::table::global_index_search_filter::matching_row_ids_for_filter;
+use crate::table::global_index_types::normalize_queryable_global_index_type;
+
+pub(super) fn intersect_include(
+ include: Option<&RoaringTreemap>,
+ mut rows: RoaringTreemap,
+) -> RoaringTreemap {
+ if let Some(include) = include {
+ rows &= include;
+ }
+ rows
+}
+
+fn row_ids(ranges: &[RowRange]) -> crate::Result<RoaringTreemap> {
+ let mut ids = RoaringTreemap::new();
+ for range in ranges {
+ let from = u64::try_from(range.from()).map_err(|_|
crate::Error::DataInvalid {
+ message: "Negative full-text candidate row ID".into(),
+ source: None,
+ })?;
+ let to = u64::try_from(range.to()).map_err(|_|
crate::Error::DataInvalid {
+ message: "Negative full-text candidate row ID".into(),
+ source: None,
+ })?;
+ ids.insert_range(from..=to);
+ }
+ Ok(ids)
+}
+
+fn ranges(ids: &RoaringTreemap) -> crate::Result<Vec<RowRange>> {
+ let mut ranges: Vec<RowRange> = Vec::new();
+ for id in ids.iter() {
+ let id = i64::try_from(id).map_err(|_| crate::Error::DataInvalid {
+ message: "Full-text candidate row ID exceeds i64".into(),
+ source: None,
+ })?;
+ match ranges.last_mut() {
+ Some(last) if last.to().checked_add(1) == Some(id) => {
+ *last = RowRange::new(last.from(), id);
+ }
+ _ => ranges.push(RowRange::new(id, id)),
+ }
+ }
+ Ok(ranges)
+}
+
+fn field_ids(predicate: &Predicate, fields: &[DataField], ids: &mut
HashSet<i32>) {
+ match predicate {
+ Predicate::Leaf { column, .. } => {
+ if let Some(id) = find_field_id_by_name(fields, column) {
+ ids.insert(id);
+ }
+ }
+ Predicate::And(children) | Predicate::Or(children) => {
+ for child in children {
+ field_ids(child, fields, ids);
+ }
+ }
+ Predicate::Not(child) => field_ids(child, fields, ids),
+ Predicate::AlwaysTrue | Predicate::AlwaysFalse => {}
+ }
+}
+
+pub(super) async fn matching_rows(
+ evaluation: &FullTextSearchEvaluation<'_>,
+ filter: &Predicate,
+ ranges: Vec<RowRange>,
+) -> crate::Result<RoaringTreemap> {
+ if ranges.is_empty() {
+ return Ok(RoaringTreemap::new());
+ }
+ let table = evaluation.table.ok_or_else(|| crate::Error::DataInvalid {
+ message: "Full-text row filtering requires table context".into(),
+ source: None,
+ })?;
+ // The caller has already selected which ranges Java permits reading. A
+ // candidate refinement must not lose rows to scalar FAST pruning again.
+ let table = table.copy_with_options(HashMap::from([(
+ "scalar-index.search-mode".into(),
+ "full".into(),
+ )]));
+ let filter = match evaluation.partition_filter {
+ Some(partition) => Predicate::and(vec![partition.clone(),
filter.clone()]),
+ None => filter.clone(),
+ };
+ matching_row_ids_for_filter(&table, &filter, Some(ranges)).await
+}
+
+pub(super) async fn matching_indexed_rows(
+ evaluation: &FullTextSearchEvaluation<'_>,
+ entries: &[IndexManifestEntry],
+ full_text_entries: &[&IndexManifestEntry],
+ filter: &Predicate,
+) -> crate::Result<RoaringTreemap> {
+ let core = CoreOptions::new(evaluation.table_options);
+ let mode = core.scalar_index_search_mode()?;
+ let covered_ranges = full_text_entries
+ .iter()
+ .map(|entry| {
+ let meta = entry.index_file.global_index_meta.as_ref().unwrap();
+ candidate_limit(meta.row_range_start, meta.row_range_end)?;
+ Ok(RowRange::new(meta.row_range_start, meta.row_range_end))
+ })
+ .collect::<crate::Result<Vec<_>>>()?;
+ let covered = row_ids(&covered_ranges)?;
+ let scalar_entries = entries
+ .iter()
+ .filter(|entry| {
+
normalize_queryable_global_index_type(&entry.index_file.index_type).is_some()
+ })
+ .cloned()
+ .collect::<Vec<_>>();
+ let mut fields = HashSet::new();
+ field_ids(filter, evaluation.schema_fields, &mut fields);
+ let detail = if mode == GlobalIndexSearchMode::Detail {
+ let table = evaluation.table.ok_or_else(|| crate::Error::DataInvalid {
+ message: "Full-text DETAIL filtering requires table
context".into(),
+ source: None,
+ })?;
+ detail_data_ranges_for_table(table, evaluation.partition_filter).await?
+ } else {
+ Vec::new()
+ };
+ let unindexed_ranges = unindexed_ranges_for_global_index_entries(
+ &scalar_entries,
+ &fields,
+ mode,
+ evaluation.next_row_id,
+ &detail,
+ |file|
normalize_queryable_global_index_type(&file.index_type).is_some(),
+ );
+ let mut unindexed = row_ids(&unindexed_ranges)? & &covered;
+ let decided = &covered - &unindexed;
+ let mut matched = RoaringTreemap::new();
+ if !decided.is_empty() {
+ // Java createForScalarFilters excludes multi-field definitions: search
+ // pre-filters evaluate individual fields rather than composite keys.
+ let scalar_entries = scalar_entries
+ .into_iter()
+ .filter(|entry| {
+ entry
+ .index_file
+ .global_index_meta
+ .as_ref()
+ .is_some_and(|meta|
meta.extra_field_ids.as_ref().is_none_or(Vec::is_empty))
+ })
+ .collect::<Vec<_>>();
+ let scanner = GlobalIndexScanner::create_with_fm_options(
+ evaluation.file_io,
+ evaluation.table_path,
+ core.global_index_thread_num()?,
+ core.btree_index_fallback_scan_max_size()?,
+ core.btree_index_data_block_cache_size()?,
+ core.bitmap_index_fallback_scan_max_size()?,
+ &scalar_entries,
+ evaluation.schema_fields,
+ if scalar_entries
+ .iter()
+ .any(|entry|
entry.index_file.index_type.eq_ignore_ascii_case("fm"))
+ {
+
crate::fm_index::FMOptions::from_options(evaluation.table_options)?.read
+ } else {
+ crate::fm_index::FMReadOptions::default()
+ },
+ )?;
+ let answer = match scanner {
+ Some(scanner) =>
scanner.matching_ranges_with_exactness(filter).await?,
+ None => None,
+ };
+ match answer {
+ Some((answer, exact)) => {
+ let candidates = row_ids(&answer)? & &decided;
+ if exact {
+ matched |= candidates;
+ } else if core.global_index_filter_refine_from_data() {
+ matched |= matching_rows(evaluation, filter,
ranges(&candidates)?).await?;
+ }
+ }
+ None if mode != GlobalIndexSearchMode::Fast => unindexed |=
decided,
+ None => {}
+ }
+ }
+ matched |= matching_rows(evaluation, filter, ranges(&unindexed)?).await?;
+ Ok(matched)
+}
diff --git a/crates/paimon/src/table/full_text_search_builder/parity_tests.rs
b/crates/paimon/src/table/full_text_search_builder/parity_tests.rs
new file mode 100644
index 00000000..fde501e5
--- /dev/null
+++ b/crates/paimon/src/table/full_text_search_builder/parity_tests.rs
@@ -0,0 +1,449 @@
+// 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.
+
+//! Real index/read coverage for the public full-text plan contract.
+use super::*;
+use crate::catalog::Identifier;
+use crate::io::FileIOBuilder;
+use crate::spec::{DataType, Datum, IntType, PredicateBuilder, Schema,
TableSchema, VarCharType};
+use crate::table::{TableCommit, TableWrite};
+use arrow_array::{ArrayRef, Int32Array};
+use arrow_schema::{DataType as ArrowType, Field, Schema as ArrowSchema};
+use std::sync::Arc;
+
+fn new_schema(partitioned: bool) -> Schema {
+ let mut schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("text", DataType::VarChar(VarCharType::string_type()))
+ .column("label", DataType::VarChar(VarCharType::string_type()))
+ .column("pt", DataType::Int(IntType::new()))
+ .options(HashMap::from([
+ ("bucket".into(), "-1".into()),
+ ("row-tracking.enabled".into(), "true".into()),
+ ("data-evolution.enabled".into(), "true".into()),
+ ("global-index.enabled".into(), "true".into()),
+ ("global-index.row-count-per-shard".into(), "4".into()),
+ ("scalar-index.search-mode".into(), "full".into()),
+ ("full-text-index.search-mode".into(), "full".into()),
+ ]));
+ if partitioned {
+ schema = schema.partition_keys(["pt"]);
+ }
+ schema.build().unwrap()
+}
+
+fn new_table(partitioned: bool) -> Table {
+ let schema = new_schema(partitioned);
+ Table::new(
+ FileIOBuilder::new("memory").build().unwrap(),
+ Identifier::new("default", "text"),
+ format!("memory:/text-parity-{}", uuid::Uuid::new_v4()),
+ TableSchema::new(0, &schema),
+ None,
+ )
+}
+
+async fn append(table: &Table, start: i32, count: i32) {
+ let schema = Arc::new(ArrowSchema::new(vec![
+ Field::new("id", ArrowType::Int32, true),
+ Field::new("text", ArrowType::Utf8, true),
+ Field::new("label", ArrowType::Utf8, true),
+ Field::new("pt", ArrowType::Int32, true),
+ ]));
+ let columns: Vec<ArrayRef> = vec![
+ Arc::new(Int32Array::from((start..start + count).collect::<Vec<_>>())),
+ Arc::new(StringArray::from(
+ (start..start + count)
+ .map(|id| if id == 3 { None } else { Some("paimon lake") })
+ .collect::<Vec<_>>(),
+ )),
+ Arc::new(StringArray::from(
+ (start..start + count)
+ .map(|id| if id % 2 == 1 { "keep" } else { "drop" })
+ .collect::<Vec<_>>(),
+ )),
+ Arc::new(Int32Array::from(
+ (start..start + count).map(|id| id / 4).collect::<Vec<_>>(),
+ )),
+ ];
+ let mut write = TableWrite::new(table, "data".into()).unwrap();
+ write
+ .write_arrow_batch(&RecordBatch::try_new(schema, columns).unwrap())
+ .await
+ .unwrap();
+ TableCommit::new(table.clone(), "data".into())
+ .commit(write.prepare_commit().await.unwrap())
+ .await
+ .unwrap();
+}
+
+async fn index(table: &Table, kind: &str, column: &str) {
+ table
+ .new_global_index_build_builder()
+ .with_index_type(kind)
+ .with_index_column(column)
+ .execute()
+ .await
+ .unwrap();
+}
+
+fn query(table: &Table) -> FullTextSearchBuilder<'_> {
+ let mut builder = table.new_full_text_search_builder();
+ builder
+ .with_query("text", r#"{"match":{"query":"paimon"}}"#)
+ .with_limit(64);
+ builder
+}
+
+fn ids(result: SearchResult) -> Vec<u64> {
+ let mut ids = result.row_ids;
+ ids.sort_unstable();
+ ids
+}
+
+#[tokio::test]
+async fn scalar_modes_and_refinement_follow_java_before_top_k() {
+ let table = new_table(false);
+ append(&table, 0, 8).await;
+ index(&table, "full-text", "text").await;
+ let pb = PredicateBuilder::new(table.schema().fields());
+ let keep = pb.equal("label", Datum::String("keep".into())).unwrap();
+ for (mode, expected) in [
+ ("fast", vec![]),
+ ("full", vec![1, 5, 7]),
+ ("detail", vec![1, 5, 7]),
+ ] {
+ let table = table.copy_with_options(HashMap::from([(
+ "scalar-index.search-mode".into(),
+ mode.into(),
+ )]));
+ assert_eq!(
+ ids(query(&table)
+ .with_filter(keep.clone())
+ .execute_scored()
+ .await
+ .unwrap()),
+ expected
+ );
+ }
+ index(&table, "btree", "label").await;
+ let fast = table.copy_with_options(HashMap::from([(
+ "scalar-index.search-mode".into(),
+ "fast".into(),
+ )]));
+ assert_eq!(
+ ids(query(&fast)
+ .with_filter(keep)
+ .execute_scored()
+ .await
+ .unwrap()),
+ vec![1, 5, 7]
+ );
+ let candidate = pb.contains("label", Datum::String("eep".into())).unwrap();
+ assert_eq!(
+ ids(query(&fast)
+ .with_filter(candidate.clone())
+ .execute_scored()
+ .await
+ .unwrap()),
+ vec![1, 5, 7]
+ );
+ let no_refine = fast.copy_with_options(HashMap::from([(
+ "global-index.filter.refine-from-data".into(),
+ "false".into(),
+ )]));
+ assert!(query(&no_refine)
+ .with_filter(candidate)
+ .execute_scored()
+ .await
+ .unwrap()
+ .is_empty());
+ let high = pb.greater_or_equal("id", Datum::Int(5)).unwrap();
+ let mut search = query(&table);
+ search.with_limit(1).with_filter(high);
+ assert_eq!(ids(search.execute_scored().await.unwrap()), vec![5]);
+}
+
+async fn user_ids(table: &Table, result: SearchResult) -> Vec<i32> {
+ let mut builder = table.new_read_builder();
+ builder.with_projection(&["id", ROW_ID_FIELD_NAME]).unwrap();
+ let plan = builder.new_scan().plan().await.unwrap();
+ let batches: Vec<RecordBatch> = builder
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ let mut mapping = HashMap::new();
+ for batch in batches {
+ let ids = batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ let row_ids = batch
+ .column(1)
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .unwrap();
+ for row in 0..batch.num_rows() {
+ mapping.insert(row_ids.value(row) as u64, ids.value(row));
+ }
+ }
+ let mut values = result
+ .row_ids
+ .iter()
+ .map(|id| mapping[id])
+ .collect::<Vec<_>>();
+ values.sort_unstable();
+ values
+}
+
+#[tokio::test]
+async fn partition_filters_apply_to_indexed_and_raw_corpora() {
+ let table = new_table(true);
+ append(&table, 0, 8).await;
+ index(&table, "full-text", "text").await;
+ append(&table, 8, 4).await;
+ let pb = PredicateBuilder::new(table.schema().fields());
+ let filter = pb.greater_or_equal("pt", Datum::Int(1)).unwrap();
+ for mode in ["full", "detail"] {
+ let table = table.copy_with_options(HashMap::from([(
+ "full-text-index.search-mode".into(),
+ mode.into(),
+ )]));
+ assert_eq!(
+ user_ids(
+ &table,
+ query(&table)
+ .with_partition_filter(filter.clone())
+ .unwrap()
+ .execute_scored()
+ .await
+ .unwrap()
+ )
+ .await,
+ (4..12).collect::<Vec<_>>()
+ );
+ let filter = Predicate::and(vec![
+ filter.clone(),
+ pb.equal("label", Datum::String("keep".into())).unwrap(),
+ ]);
+ assert_eq!(
+ user_ids(
+ &table,
+ query(&table)
+ .with_filter(filter)
+ .execute_scored()
+ .await
+ .unwrap()
+ )
+ .await,
+ vec![5, 7, 9, 11]
+ );
+ }
+}
+
+#[tokio::test]
+async fn scan_read_pin_snapshot_and_validate_plan_origin() {
+ let table = new_table(false);
+ append(&table, 0, 8).await;
+ index(&table, "full-text", "text").await;
+ let builder = query(&table);
+ let scan = builder.new_scan().unwrap();
+ let read = builder.new_read().unwrap();
+ let plan = scan.scan().await.unwrap();
+ assert_eq!(plan.snapshot_id(), Some(2));
+ append(&table, 8, 4).await;
+ assert_eq!(
+ ids(read.read(plan.clone()).await.unwrap()),
+ vec![0, 1, 2, 4, 5, 6, 7]
+ );
+ assert_eq!(
+ ids(query(&table).execute_scored().await.unwrap()),
+ vec![0, 1, 2, 4, 5, 6, 7, 8, 9, 10, 11]
+ );
+ let pb = PredicateBuilder::new(table.schema().fields());
+ let mut other = query(&table);
+ other.with_filter(pb.equal("id", Datum::Int(1)).unwrap());
+ assert!(other
+ .new_read()
+ .unwrap()
+ .read(plan.clone())
+ .await
+ .unwrap_err()
+ .to_string()
+ .contains("different table, column or filter"));
+ let foreign = new_table(false);
+ assert!(query(&foreign)
+ .new_read()
+ .unwrap()
+ .read(plan)
+ .await
+ .is_err());
+}
+
+#[tokio::test]
+async fn empty_plans_stay_empty_after_append_and_validate_query_first() {
+ let table = new_table(false);
+ let builder = query(&table);
+ let plan = builder.new_scan().unwrap().scan().await.unwrap();
+ assert_eq!(plan.snapshot_id(), None);
+ append(&table, 0, 4).await;
+ index(&table, "full-text", "text").await;
+ assert!(builder
+ .new_read()
+ .unwrap()
+ .read(plan)
+ .await
+ .unwrap()
+ .is_empty());
+ let empty = new_table(false);
+ assert!(query(&empty)
+ .with_limit(0)
+ .execute_scored()
+ .await
+ .unwrap_err()
+ .to_string()
+ .contains("positive"));
+ assert!(query(&empty)
+ .with_query("missing", r#"{"match":{"query":"paimon"}}"#)
+ .new_scan()
+ .is_err());
+ assert!(query(&empty)
+ .with_query("id", r#"{"match":{"query":"paimon"}}"#)
+ .new_scan()
+ .is_err());
+}
+
+#[tokio::test]
+async fn raw_filter_keeps_bm25_statistics_of_the_whole_corpus() {
+ let table = new_table(false);
+ append(&table, 0, 4).await;
+ index(&table, "full-text", "text").await;
+ append(&table, 4, 8).await;
+ let unfiltered = query(&table).execute_scored().await.unwrap();
+ let pb = PredicateBuilder::new(table.schema().fields());
+ let filtered = query(&table)
+ .with_filter(pb.equal("label", Datum::String("keep".into())).unwrap())
+ .execute_scored()
+ .await
+ .unwrap();
+ assert_eq!(ids(filtered.clone()), vec![1, 5, 7, 9, 11]);
+ let scores: HashMap<_, _> = unfiltered
+ .row_ids
+ .into_iter()
+ .zip(unfiltered.scores)
+ .collect();
+ for (id, score) in filtered.row_ids.into_iter().zip(filtered.scores) {
+ assert_eq!(score, scores[&id]);
+ }
+}
+
+#[tokio::test]
+async fn snapshot_pinning_preserves_schema_only_renames() {
+ use crate::catalog::{Catalog, FileSystemCatalog};
+ use crate::common::Options;
+ use crate::spec::SchemaChange;
+
+ let catalog = FileSystemCatalog::new(Options::from_map(HashMap::from([(
+ "warehouse".into(),
+ format!("memory:/text-schema-rename-{}", uuid::Uuid::new_v4()),
+ )])))
+ .unwrap();
+ catalog
+ .create_database("default", false, HashMap::new())
+ .await
+ .unwrap();
+ let identifier = Identifier::new("default", "text");
+ catalog
+ .create_table(&identifier, new_schema(false), false)
+ .await
+ .unwrap();
+ let table = catalog.get_table(&identifier).await.unwrap();
+ append(&table, 0, 4).await;
+ index(&table, "full-text", "text").await;
+ append(&table, 4, 4).await;
+ catalog
+ .alter_table(
+ &identifier,
+ vec![SchemaChange::rename_column("text".into(), "body".into())],
+ false,
+ )
+ .await
+ .unwrap();
+ let table = catalog.get_table(&identifier).await.unwrap();
+ for mode in ["full", "detail"] {
+ let table = table.copy_with_options(HashMap::from([(
+ "full-text-index.search-mode".into(),
+ mode.into(),
+ )]));
+ let mut builder = table.new_full_text_search_builder();
+ builder
+ .with_query("body", r#"{"match":{"query":"paimon"}}"#)
+ .with_limit(64);
+ assert_eq!(
+ ids(builder.execute_scored().await.unwrap()),
+ vec![0, 1, 2, 4, 5, 6, 7]
+ );
+ let plan = builder.new_scan().unwrap().scan().await.unwrap();
+ assert_eq!(plan.table.schema().id(), table.schema().id());
+ assert_eq!(
+ ids(builder.new_read().unwrap().read(plan).await.unwrap()),
+ vec![0, 1, 2, 4, 5, 6, 7]
+ );
+ }
+}
+
+#[tokio::test]
+async fn scalar_search_pre_filters_exclude_composite_indexes() {
+ let table = new_table(false);
+ append(&table, 0, 8).await;
+ index(&table, "full-text", "text").await;
+ table
+ .new_global_index_build_builder()
+ .with_index_type("btree")
+ .with_index_columns(&["label", "id"])
+ .execute()
+ .await
+ .unwrap();
+ let pb = PredicateBuilder::new(table.schema().fields());
+ let predicate = Predicate::and(vec![
+ pb.equal("label", Datum::String("keep".into())).unwrap(),
+ pb.greater_than("id", Datum::Int(0)).unwrap(),
+ ]);
+ for (mode, expected) in [
+ ("fast", vec![]),
+ ("full", vec![1, 5, 7]),
+ ("detail", vec![1, 5, 7]),
+ ] {
+ let table = table.copy_with_options(HashMap::from([(
+ "scalar-index.search-mode".into(),
+ mode.into(),
+ )]));
+ assert_eq!(
+ ids(query(&table)
+ .with_filter(predicate.clone())
+ .execute_scored()
+ .await
+ .unwrap()),
+ expected
+ );
+ }
+}
diff --git a/crates/paimon/src/table/full_text_search_builder/read.rs
b/crates/paimon/src/table/full_text_search_builder/read.rs
new file mode 100644
index 00000000..3a8b2169
--- /dev/null
+++ b/crates/paimon/src/table/full_text_search_builder/read.rs
@@ -0,0 +1,89 @@
+// 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.
+
+//! Executes full-text queries over a snapshot-scoped plan.
+
+use super::scan::PlanContext;
+use super::*;
+
+/// An owned reader can execute queries over plans from the same table, column
+/// and filter. Changing a query or limit does not require another scan.
+pub struct FullTextRead {
+ table: Table,
+ context: PlanContext,
+ search: FullTextSearch,
+}
+
+impl FullTextRead {
+ pub(super) fn new(builder: &FullTextSearchBuilder<'_>) ->
crate::Result<Self> {
+
CoreOptions::new(builder.table.schema().options()).ensure_read_authorized()?;
+ let limit = builder.limit.filter(|limit| *limit > 0).ok_or_else(|| {
+ crate::Error::ConfigInvalid {
+ message: "Limit must be positive, set via with_limit()".into(),
+ }
+ })?;
+ let context = PlanContext::new(builder)?;
+ let query = builder
+ .query_text
+ .as_deref()
+ .ok_or_else(|| crate::Error::ConfigInvalid {
+ message: "Query text must be set via with_query() or
with_query_text()".into(),
+ })?;
+ let query = if builder.query_is_dsl {
+ query.to_string()
+ } else {
+ normalize_query_text(query, &context.column)?
+ };
+ let search = FullTextSearch::new(query, limit,
context.column.clone())?;
+ Ok(Self {
+ table: builder.table.clone(),
+ context,
+ search,
+ })
+ }
+
+ pub async fn read(&self, plan: FullTextScanPlan) ->
crate::Result<SearchResult> {
+
CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?;
+
CoreOptions::new(plan.table.schema().options()).ensure_read_authorized()?;
+ if self.context != plan.context {
+ return Err(crate::Error::DataInvalid {
+ message: "Full-text plan belongs to a different table, column
or filter".into(),
+ source: None,
+ });
+ }
+ if plan.snapshot_id.is_none() {
+ return Ok(SearchResult::empty());
+ }
+ let mut search = self.search.clone();
+ search.include_row_ids = self.context.include_row_ids.clone();
+ evaluate_full_text_search(
+ FullTextSearchEvaluation {
+ table: Some(&plan.table),
+ file_io: plan.table.file_io(),
+ table_path: plan.table.location(),
+ table_options: self.table.schema().options(),
+ schema_fields: self.table.schema().fields(),
+ next_row_id: plan.next_row_id,
+ partition_filter: plan.partition_filter.as_ref(),
+ row_filter: plan.row_filter.as_ref(),
+ },
+ &plan.entries,
+ &search,
+ )
+ .await
+ }
+}
diff --git a/crates/paimon/src/table/full_text_search_builder/scan.rs
b/crates/paimon/src/table/full_text_search_builder/scan.rs
new file mode 100644
index 00000000..8e87a3cd
--- /dev/null
+++ b/crates/paimon/src/table/full_text_search_builder/scan.rs
@@ -0,0 +1,183 @@
+// 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.
+
+//! Snapshot-pinned, query-independent full-text planning.
+
+use super::*;
+use crate::spec::Predicate;
+use crate::table::bucket_filter::split_partition_and_data_predicates;
+use crate::table::partition_filter::PartitionFilter;
+
+#[derive(Clone, PartialEq)]
+pub(super) struct PlanContext {
+ location: String,
+ branch: String,
+ pub(super) column: String,
+ pub(super) filter: Option<Predicate>,
+ pub(super) include_row_ids: Option<RoaringTreemap>,
+}
+
+impl PlanContext {
+ pub(super) fn new(builder: &FullTextSearchBuilder<'_>) ->
crate::Result<Self> {
+
CoreOptions::new(builder.table.schema().options()).ensure_read_authorized()?;
+ let column = builder
+ .text_column
+ .as_deref()
+ .ok_or_else(|| crate::Error::ConfigInvalid {
+ message: "Text column must be set via with_query() or
with_text_column()".into(),
+ })?;
+ let field = builder
+ .table
+ .schema()
+ .fields()
+ .iter()
+ .find(|field| field.name() == column)
+ .ok_or_else(|| crate::Error::ConfigInvalid {
+ message: format!("Text column '{column}' does not exist"),
+ })?;
+ if !matches!(
+ field.data_type(),
+ crate::spec::DataType::Char(_) | crate::spec::DataType::VarChar(_)
+ ) {
+ return Err(crate::Error::ConfigInvalid {
+ message: format!("Full-text column '{column}' must be a
character string"),
+ });
+ }
+ if
resolves_to_pk_full_text_path(&builder.table.schema().core_options(), column) {
+ return Err(crate::Error::DataInvalid { message: "primary-key
full-text search does not produce global row ids; use the materialized read
(execute_read) instead".into(), source: None });
+ }
+ Ok(Self {
+ location: builder.table.location().trim_end_matches('/').into(),
+ branch: builder.table.branch().into(),
+ column: column.into(),
+ filter: builder.filter.clone(),
+ include_row_ids: builder.include_row_ids.clone(),
+ })
+ }
+}
+
+/// One selected snapshot and its partition-pruned index entries. A plan
remains
+/// usable after its scan is dropped, and does not re-resolve the latest
snapshot.
+#[derive(Clone)]
+pub struct FullTextScanPlan {
+ pub(super) context: PlanContext,
+ pub(super) table: Table,
+ pub(super) entries: Vec<IndexManifestEntry>,
+ pub(super) next_row_id: Option<i64>,
+ pub(super) snapshot_id: Option<i64>,
+ pub(super) partition_filter: Option<Predicate>,
+ pub(super) row_filter: Option<Predicate>,
+}
+
+impl FullTextScanPlan {
+ pub fn snapshot_id(&self) -> Option<i64> {
+ self.snapshot_id
+ }
+}
+
+/// Full-text planning follows Java FullTextScan: select one snapshot and prune
+/// index manifest entries by partition. Queries and limits belong to the
reader.
+pub struct FullTextScan {
+ table: Table,
+ context: PlanContext,
+}
+
+impl FullTextScan {
+ pub(super) fn new(builder: &FullTextSearchBuilder<'_>) ->
crate::Result<Self> {
+ Ok(Self {
+ table: builder.table.clone(),
+ context: PlanContext::new(builder)?,
+ })
+ }
+
+ pub async fn scan(&self) -> crate::Result<FullTextScanPlan> {
+
CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?;
+ let mut plan = FullTextScanPlan {
+ context: self.context.clone(),
+ table: self.table.clone(),
+ entries: Vec::new(),
+ next_row_id: None,
+ snapshot_id: None,
+ partition_filter: None,
+ row_filter: None,
+ };
+ let mut partition_pruner = None;
+ if let Some(filter) = &self.context.filter {
+ let (partition, data) = split_partition_and_data_predicates(
+ filter.clone(),
+ self.table.schema().fields(),
+ self.table.schema().partition_keys(),
+ );
+ if let Some(partition) = partition {
+ let fields = self.table.schema().partition_fields();
+ partition_pruner =
+ Some(PartitionFilter::from_predicate(partition.clone(),
&fields));
+ let mapping = self
+ .table
+ .schema()
+ .partition_keys()
+ .iter()
+ .map(|key| {
+ self.table
+ .schema()
+ .fields()
+ .iter()
+ .position(|field| field.name() == key)
+ })
+ .collect::<Vec<_>>();
+ plan.partition_filter =
+ Some(partition.remap_field_index(&mapping).ok_or_else(|| {
+ crate::Error::DataInvalid {
+ message:
+ "Cannot bind full-text partition predicate to
the table schema"
+ .into(),
+ source: None,
+ }
+ })?);
+ }
+ plan.row_filter = (!data.is_empty()).then(|| Predicate::and(data));
+ }
+ if plan.row_filter.is_some() &&
!self.table.schema().core_options().data_evolution_enabled()
+ {
+ return Err(crate::Error::Unsupported {
+ message: "Full-text row filters require a data-evolution
table".into(),
+ });
+ }
+ let Some(snapshot) =
crate::table::time_travel::resolve_snapshot(&self.table).await? else {
+ return Ok(plan);
+ };
+ plan.snapshot_id = Some(snapshot.id());
+ plan.next_row_id = snapshot.next_row_id();
+ // Pin the data without changing the schema already bound by the query.
+ // A schema-only rename does not create a new data snapshot.
+ plan.table = self.table.copy_with_pinned_snapshot(&snapshot);
+ if let Some(name) = snapshot.index_manifest() {
+ let path = self.table.snapshot_manager().manifest_path(name);
+ plan.entries = IndexManifest::read(self.table.file_io(),
&path).await?;
+ }
+ if let Some(partition) = partition_pruner {
+ let mut entries = Vec::new();
+ for entry in plan.entries {
+ if partition.matches_entry(&entry.partition)? {
+ entries.push(entry);
+ }
+ }
+ plan.entries = entries;
+ }
+ Ok(plan)
+ }
+}
diff --git a/crates/paimon/src/table/global_index_search_filter.rs
b/crates/paimon/src/table/global_index_search_filter.rs
new file mode 100644
index 00000000..423bd867
--- /dev/null
+++ b/crates/paimon/src/table/global_index_search_filter.rs
@@ -0,0 +1,77 @@
+// 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-ID filtering shared by data-evolution index searches.
+
+use crate::spec::{Predicate, ROW_ID_FIELD_NAME};
+use crate::table::{RowRange, Table};
+use arrow_array::{Array, Int64Array};
+use futures::TryStreamExt;
+use roaring::RoaringTreemap;
+
+pub(super) async fn matching_row_ids_for_filter(
+ table: &Table,
+ filter: &Predicate,
+ ranges: Option<Vec<RowRange>>,
+) -> crate::Result<RoaringTreemap> {
+ let mut read_builder = table.new_read_builder();
+ read_builder
+ .with_projection(&[ROW_ID_FIELD_NAME])?
+ .with_filter(filter.clone());
+ if let Some(ranges) = ranges {
+ read_builder.with_row_ranges(ranges);
+ }
+ let plan = read_builder.new_scan().plan().await?;
+ let read = read_builder.new_read()?;
+ let mut stream = read.to_arrow(plan.splits())?;
+ let mut row_ids = RoaringTreemap::new();
+ while let Some(batch) = stream.try_next().await? {
+ let index =
+ batch
+ .schema()
+ .index_of(ROW_ID_FIELD_NAME)
+ .map_err(|_| crate::Error::DataInvalid {
+ message: format!("global index row filter read is missing
{ROW_ID_FIELD_NAME}"),
+ source: None,
+ })?;
+ let values = batch
+ .column(index)
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .ok_or_else(|| crate::Error::DataInvalid {
+ message: format!("global index row filter {ROW_ID_FIELD_NAME}
column is not Int64"),
+ source: None,
+ })?;
+ for row in 0..values.len() {
+ if values.is_null(row) {
+ return Err(crate::Error::DataInvalid {
+ message: format!("global index row filter produced a null
{ROW_ID_FIELD_NAME}"),
+ source: None,
+ });
+ }
+ let row_id = values.value(row);
+ let row_id = u64::try_from(row_id).map_err(|_|
crate::Error::DataInvalid {
+ message: format!(
+ "global index row filter produced a negative
{ROW_ID_FIELD_NAME}: {row_id}"
+ ),
+ source: None,
+ })?;
+ row_ids.insert(row_id);
+ }
+ }
+ Ok(row_ids)
+}
diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs
index 3877bd6f..11a902ed 100644
--- a/crates/paimon/src/table/mod.rs
+++ b/crates/paimon/src/table/mod.rs
@@ -78,6 +78,7 @@ mod global_index_build_builder;
pub(crate) mod global_index_build_common;
mod global_index_drop_builder;
pub(crate) mod global_index_scanner;
+mod global_index_search_filter;
mod global_index_types;
mod hybrid_search_builder;
mod incremental_scan;
@@ -189,7 +190,9 @@ pub use format_partition::{
pub use format_partition_stats::FormatTablePartitionStatsCollector;
pub use format_partition_truncate::FormatTableTruncator;
#[cfg(feature = "fulltext")]
-pub use full_text_search_builder::FullTextSearchBuilder;
+pub use full_text_search_builder::{
+ FullTextRead, FullTextScan, FullTextScanPlan, FullTextSearchBuilder,
+};
use futures::stream::BoxStream;
pub use global_index_build_builder::GlobalIndexBuildBuilder;
pub use global_index_drop_builder::GlobalIndexDropBuilder;
diff --git a/docs/src/sql.md b/docs/src/sql.md
index e017bedb..120b8a33 100644
--- a/docs/src/sql.md
+++ b/docs/src/sql.md
@@ -2129,6 +2129,40 @@ The function searches across all Tantivy full-text index
files for the target co
Build the index files with `CALL sys.create_global_index(..., index_type =>
'full-text')` (see [create_global_index](#create_global_index)), or with Java
Paimon; both write the same format.
+### Rust full-text builder
+
+The core builder accepts the structured full-text query DSL used by Java and
+PyPaimon. Scalar and partition predicates are applied before the final Top-K.
+
+```rust,ignore
+let predicate = PredicateBuilder::new(table.schema().fields())
+ .equal("category", Datum::String("news".into()))?;
+let mut builder = table.new_full_text_search_builder();
+builder
+ .with_query("content", r#"{"match":{"query":"paimon"}}"#)
+ .with_filter(predicate)
+ .with_limit(10);
+let plan = builder.new_scan()?.scan().await?;
+let result = builder.new_read()?.read(plan).await?;
+```
+
+`FullTextScanPlan` owns a pinned snapshot and can outlive its scan. Readers
+for the same table, branch, column and filter can reuse the plan with different
+queries or limits. Pinning preserves the query's bound schema, including
+schema-only column renames. `execute_scored()` performs both stages directly.
+
+For Data Evolution, full-text and scalar `fast`, `full`, and `detail` modes
+follow Java's coverage rules. Inexact scalar candidates are read back only
+when `global-index.filter.refine-from-data=true`. Composite scalar indexes
+are excluded from search pre-filters, as in Java. Filtering preserves the
+unfiltered corpus for BM25 scores; Boolean and boosted queries collect leaf
+candidates before applying the final limit. Index shard searches use
+`global-index.thread-num` workers.
+
+The existing `with_text_column()` / `with_query_text()` API continues to
+accept plain text for the SQL table function. Full-text row filters require
+Data Evolution; primary-key materialized reads use `execute_read()`.
+
## Time Travel
