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 7baef42f feat(core): support full-text prefilters on non-DE append
tables (#1176)
7baef42f is described below
commit 7baef42fbeb97353036036b1bb71a1f57510e00b
Author: QuakeWang <[email protected]>
AuthorDate: Thu Oct 22 14:51:04 2026 +0800
feat(core): support full-text prefilters on non-DE append tables (#1176)
---
.../table/full_text_search_builder/append_tests.rs | 385 +++++++++++++++++++++
.../src/table/full_text_search_builder/filter.rs | 8 +-
.../table/full_text_search_builder/parity_tests.rs | 3 +
.../src/table/full_text_search_builder/scan.rs | 37 +-
.../paimon/src/table/global_index_search_filter.rs | 2 +-
docs/src/sql.md | 15 +-
6 files changed, 436 insertions(+), 14 deletions(-)
diff --git a/crates/paimon/src/table/full_text_search_builder/append_tests.rs
b/crates/paimon/src/table/full_text_search_builder/append_tests.rs
new file mode 100644
index 00000000..cd810e20
--- /dev/null
+++ b/crates/paimon/src/table/full_text_search_builder/append_tests.rs
@@ -0,0 +1,385 @@
+// 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.
+
+//! Non-data-evolution append search uses real Parquet data and global indexes.
+
+use super::*;
+use crate::deletion_vector::DeletionVector;
+use crate::spec::{DeletionVectorMeta, IndexFileMeta};
+use crate::table::CommitMessage;
+use roaring::RoaringBitmap;
+
+fn append_table(partitioned: bool) -> Table {
+ new_table(partitioned).copy_with_options(HashMap::from([
+ ("data-evolution.enabled".into(), "false".into()),
+ ("file.format".into(), "parquet".into()),
+ ("full-text-index.search-mode".into(), "fast".into()),
+ ("scalar-index.search-mode".into(), "fast".into()),
+ ]))
+}
+
+fn keep_filter(table: &Table) -> Predicate {
+ PredicateBuilder::new(table.schema().fields())
+ .equal("label", Datum::String("keep".into()))
+ .unwrap()
+}
+
+#[tokio::test]
+async fn scalar_modes_preserve_missing_and_partial_coverage() {
+ let table = append_table(false);
+ append(&table, 0, 4).await;
+ index(&table, "full-text", "text").await;
+ let filter = keep_filter(&table);
+ for (mode, expected) in [("fast", vec![]), ("full", vec![1]), ("detail",
vec![1])] {
+ let table = table.copy_with_options(HashMap::from([(
+ "scalar-index.search-mode".into(),
+ mode.into(),
+ )]));
+ assert_eq!(
+ ids(query(&table)
+ .with_filter(filter.clone())
+ .execute_scored()
+ .await
+ .unwrap()),
+ expected,
+ "missing scalar coverage, mode={mode}",
+ );
+ }
+
+ index(&table, "btree", "label").await;
+ append(&table, 4, 4).await;
+ index(&table, "full-text", "text").await;
+ append(&table, 8, 4).await;
+ for (mode, expected) in [
+ ("fast", vec![1]),
+ ("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(filter.clone())
+ .execute_scored()
+ .await
+ .unwrap()),
+ expected,
+ "partial scalar coverage, mode={mode}",
+ );
+ }
+}
+
+#[tokio::test]
+async fn candidate_refinement_keeps_rust_defaults_and_original_scores() {
+ let table = append_table(false);
+ append_labels(
+ &table,
+ 0,
+ &[
+ None,
+ Some("drop"),
+ Some("keep"),
+ Some("drop"),
+ Some("keep"),
+ Some("keeper"),
+ Some("other"),
+ Some("keep"),
+ ],
+ )
+ .await;
+ index(&table, "full-text", "text").await;
+ index(&table, "btree", "label").await;
+ assert!(table
+ .schema()
+ .core_options()
+ .global_index_filter_refine_from_data());
+ let pb = PredicateBuilder::new(table.schema().fields());
+ let filter = Predicate::and(vec![
+ pb.contains("label", Datum::String("eep".into())).unwrap(),
+ pb.greater_or_equal("id", Datum::Int(5)).unwrap(),
+ ]);
+ let all = query(&table)
+ .execute_scored()
+ .await
+ .unwrap()
+ .into_row_ids()
+ .unwrap();
+ let filtered = query(&table)
+ .with_filter(filter.clone())
+ .with_limit(2)
+ .execute_scored()
+ .await
+ .unwrap();
+ assert_eq!(ids(filtered.clone()), vec![5, 7]);
+ let filtered = filtered.into_row_ids().unwrap();
+ for (id, score) in filtered.row_ids.iter().zip(&filtered.scores) {
+ let position = all.row_ids.iter().position(|row| row == id).unwrap();
+ assert_eq!(*score, all.scores[position]);
+ }
+
+ let no_refine = table.copy_with_options(HashMap::from([(
+ "global-index.filter.refine-from-data".into(),
+ "false".into(),
+ )]));
+ assert!(query(&no_refine)
+ .with_filter(filter)
+ .execute_scored()
+ .await
+ .unwrap()
+ .is_empty());
+ assert!(query(&table)
+ .with_filter(pb.equal("label",
Datum::String("missing".into())).unwrap())
+ .execute_scored()
+ .await
+ .unwrap()
+ .is_empty());
+}
+
+#[tokio::test]
+async fn partition_scalar_and_include_filters_precede_top_k() {
+ for kind in ["btree", "bitmap"] {
+ let table = append_table(true);
+ append(&table, 0, 8).await;
+ index(&table, "full-text", "text").await;
+ index(&table, kind, "label").await;
+ append(&table, 8, 4).await;
+ let pb = PredicateBuilder::new(table.schema().fields());
+ let filter = Predicate::and(vec![
+ pb.greater_or_equal("pt", Datum::Int(1)).unwrap(),
+ keep_filter(&table),
+ ]);
+ let mut builder = query(&table);
+ builder.with_filter(filter).with_limit(2);
+ let result = builder.execute_scored().await.unwrap();
+ assert_eq!(user_ids(&table, result.clone()).await, vec![5, 7]);
+ // Partition commit order assigns global IDs independently of user IDs.
+ let last = *result.row_ids().unwrap().row_ids.iter().max().unwrap();
+ builder.with_include_row_ids([last, 9].into_iter().collect());
+ assert_eq!(
+ user_ids(&table, builder.execute_scored().await.unwrap()).await,
+ vec![7],
+ );
+ builder.with_include_row_ids(RoaringTreemap::new());
+ assert!(builder.execute_scored().await.unwrap().is_empty());
+ }
+}
+
+async fn delete_row(table: &Table, row_id: i64) {
+ let plan = table.new_read_builder().new_scan().plan().await.unwrap();
+ let (split, file) = plan
+ .splits()
+ .iter()
+ .flat_map(|split| split.data_files().iter().map(move |file| (split,
file)))
+ .find(|(_, file)| {
+ let first = file.first_row_id.unwrap();
+ first <= row_id && row_id < first + file.row_count
+ })
+ .unwrap();
+ let position = u32::try_from(row_id - file.first_row_id.unwrap()).unwrap();
+ let dv = DeletionVector::from_bitmap(RoaringBitmap::from_iter([position]));
+ let serialized = dv.serialize_to_bytes().unwrap();
+ let length = i32::from_be_bytes(serialized[..4].try_into().unwrap());
+ let mut bytes = vec![1];
+ bytes.extend_from_slice(&serialized);
+ let file_size = bytes.len() as i64;
+ let file_name = format!("index-{}-1", uuid::Uuid::new_v4());
+ table
+ .file_io()
+ .new_output(&format!("{}/index/{file_name}", table.location()))
+ .unwrap()
+ .write(bytes::Bytes::from(bytes))
+ .await
+ .unwrap();
+ let mut message = CommitMessage::new(
+ split.partition().to_serialized_bytes(),
+ split.bucket(),
+ vec![],
+ );
+ message.new_index_files.push(IndexFileMeta {
+ index_type: "DELETION_VECTORS".into(),
+ file_name,
+ file_size,
+ row_count: 1,
+ deletion_vectors_ranges: Some(indexmap::IndexMap::from([(
+ file.file_name.clone(),
+ DeletionVectorMeta {
+ offset: 1,
+ length,
+ cardinality: Some(1),
+ },
+ )])),
+ external_path: None,
+ global_index_meta: None,
+ });
+ TableCommit::new(table.clone(), "delete".into())
+ .commit(vec![message])
+ .await
+ .unwrap();
+}
+
+#[tokio::test]
+async fn
deletion_vectors_filter_exact_index_hits_before_top_k_and_pin_snapshot() {
+ let table = append_table(false);
+ append(&table, 0, 4).await;
+ append(&table, 4, 4).await;
+ index(&table, "full-text", "text").await;
+ index(&table, "btree", "label").await;
+ let table = table.copy_with_options(HashMap::from([(
+ "deletion-vectors.enabled".into(),
+ "true".into(),
+ )]));
+ let mut builder = query(&table);
+ builder.with_filter(keep_filter(&table)).with_limit(2);
+ let read = builder.new_read().unwrap();
+ let plan = builder.new_scan().unwrap().scan().await.unwrap();
+ delete_row(&table, 5).await;
+
+ assert_eq!(ids(read.read(plan).await.unwrap()), vec![1, 5]);
+ let result = builder.execute_scored().await.unwrap();
+ assert_eq!(ids(result.clone()), vec![1, 7]);
+ let batches: Vec<RecordBatch> = result
+ .new_read_builder()
+ .with_projection(&["id"])
+ .read()
+ .await
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ let values = batches
+ .iter()
+ .flat_map(|batch| {
+ batch
+ .column_by_name("id")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap()
+ .values()
+ .iter()
+ .copied()
+ })
+ .collect::<Vec<_>>();
+ assert_eq!(values, vec![1, 7]);
+}
+
+#[tokio::test]
+async fn removed_files_do_not_supply_exact_index_hits() {
+ let table = append_table(false);
+ append(&table, 0, 4).await;
+ append(&table, 4, 4).await;
+ index(&table, "full-text", "text").await;
+ index(&table, "btree", "label").await;
+ let mut builder = query(&table);
+ builder.with_filter(keep_filter(&table)).with_limit(1);
+ let read = builder.new_read().unwrap();
+ let plan = builder.new_scan().unwrap().scan().await.unwrap();
+ let data = table.new_read_builder().new_scan().plan().await.unwrap();
+ let split = data
+ .splits()
+ .iter()
+ .find(|split| {
+ split
+ .data_files()
+ .iter()
+ .any(|file| file.first_row_id == Some(0))
+ })
+ .unwrap();
+ let mut message = CommitMessage::new(
+ split.partition().to_serialized_bytes(),
+ split.bucket(),
+ vec![],
+ );
+ message.deleted_files = split
+ .data_files()
+ .iter()
+ .filter(|file| file.first_row_id == Some(0))
+ .cloned()
+ .collect();
+ TableCommit::new(table.clone(), "remove".into())
+ .commit(vec![message])
+ .await
+ .unwrap();
+ assert_eq!(ids(read.read(plan).await.unwrap()), vec![1]);
+ assert_eq!(ids(builder.execute_scored().await.unwrap()), vec![5]);
+}
+
+#[tokio::test]
+async fn empty_filtered_plans_stay_empty_after_index_build() {
+ let table = append_table(false);
+ let mut builder = query(&table);
+ builder.with_filter(keep_filter(&table));
+ let read = builder.new_read().unwrap();
+ 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;
+ index(&table, "btree", "label").await;
+ assert!(read.read(plan).await.unwrap().is_empty());
+ assert_eq!(ids(builder.execute_scored().await.unwrap()), vec![1]);
+}
+
+#[tokio::test]
+async fn unsupported_append_filter_combinations_fail_before_empty_snapshot() {
+ let table = append_table(false);
+ for (option, value, expected) in [
+ (
+ "row-tracking.enabled",
+ "false",
+ "row-tracked Parquet append",
+ ),
+ ("file.format", "orc", "row-tracked Parquet append"),
+ ("file.format", "avro", "row-tracked Parquet append"),
+ ("full-text-index.search-mode", "full", "FAST"),
+ ("full-text-index.search-mode", "detail", "FAST"),
+ ("query-auth.enabled", "true", "query"),
+ ] {
+ let table = table.copy_with_options(HashMap::from([(option.into(),
value.into())]));
+ let error = query(&table)
+ .with_filter(keep_filter(&table))
+ .execute_scored()
+ .await
+ .unwrap_err();
+ assert!(
+ error.to_string().contains(expected),
+ "{option}={value}: {error}"
+ );
+ }
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("text", DataType::VarChar(VarCharType::string_type()))
+ .column("label", DataType::VarChar(VarCharType::string_type()))
+ .primary_key(["id"])
+ .build()
+ .unwrap();
+ let pk = Table::new(
+ table.file_io().clone(),
+ table.identifier().clone(),
+ table.location().into(),
+ TableSchema::new(0, &schema),
+ None,
+ );
+ assert!(query(&pk)
+ .with_filter(keep_filter(&pk))
+ .execute_scored()
+ .await
+ .unwrap_err()
+ .to_string()
+ .contains("row-tracked Parquet append"));
+}
diff --git a/crates/paimon/src/table/full_text_search_builder/filter.rs
b/crates/paimon/src/table/full_text_search_builder/filter.rs
index eef7a1a9..bfca94b4 100644
--- a/crates/paimon/src/table/full_text_search_builder/filter.rs
+++ b/crates/paimon/src/table/full_text_search_builder/filter.rs
@@ -195,7 +195,13 @@ pub(super) async fn matching_indexed_rows(
Some((answer, exact)) => {
let candidates = row_ids(&answer)? & &decided;
if exact {
- matched |= candidates;
+ matched |= if core.data_evolution_enabled() {
+ candidates
+ } else {
+ // An exact index match does not establish snapshot
visibility.
+ matching_rows(evaluation, &Predicate::AlwaysTrue,
ranges(&candidates)?)
+ .await?
+ };
} else if core.global_index_filter_refine_from_data() {
matched |= matching_rows(evaluation, filter,
ranges(&candidates)?).await?;
}
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
index 03fc09a1..90b365d6 100644
--- a/crates/paimon/src/table/full_text_search_builder/parity_tests.rs
+++ b/crates/paimon/src/table/full_text_search_builder/parity_tests.rs
@@ -25,6 +25,9 @@ use arrow_array::{ArrayRef, Int32Array};
use arrow_schema::{DataType as ArrowType, Field, Schema as ArrowSchema};
use std::sync::Arc;
+#[path = "append_tests.rs"]
+mod append_tests;
+
fn new_schema(partitioned: bool) -> Schema {
let mut schema = Schema::builder()
.column("id", DataType::Int(IntType::new()))
diff --git a/crates/paimon/src/table/full_text_search_builder/scan.rs
b/crates/paimon/src/table/full_text_search_builder/scan.rs
index d26a1a5a..1971bc33 100644
--- a/crates/paimon/src/table/full_text_search_builder/scan.rs
+++ b/crates/paimon/src/table/full_text_search_builder/scan.rs
@@ -32,7 +32,8 @@ pub(super) struct PlanContext {
impl PlanContext {
pub(super) fn new(builder: &FullTextSearchBuilder<'_>) ->
crate::Result<Self> {
-
CoreOptions::new(builder.table.schema().options()).ensure_read_authorized()?;
+ let core = builder.table.schema().core_options();
+ core.ensure_read_authorized()?;
let column = builder
.text_column
.as_deref()
@@ -56,9 +57,31 @@ impl PlanContext {
message: format!("Full-text column '{column}' must be a
character string"),
});
}
+ let has_row_filter = builder.filter.as_ref().is_some_and(|filter| {
+ let (_, data) = split_partition_and_data_predicates(
+ filter.clone(),
+ builder.table.schema().fields(),
+ builder.table.schema().partition_keys(),
+ );
+ !data.is_empty()
+ });
+ if has_row_filter && !core.data_evolution_enabled() {
+ if !builder.table.schema().primary_keys().is_empty()
+ || !core.row_tracking_enabled()
+ || core.file_format() != "parquet"
+ {
+ return Err(crate::Error::Unsupported {
+ message: "Non-data-evolution full-text row filters require
a row-tracked Parquet append table".into(),
+ });
+ }
+ if core.full_text_index_search_mode()? !=
GlobalIndexSearchMode::Fast {
+ return Err(crate::Error::Unsupported {
+ message: "Non-data-evolution full-text row filters support
only the FAST full-text index search mode".into(),
+ });
+ }
+ }
let primary_key_field_id =
-
resolves_to_pk_full_text_path(&builder.table.schema().core_options(), column)
- .then_some(field.id());
+ resolves_to_pk_full_text_path(&core, column).then_some(field.id());
if primary_key_field_id.is_some() && builder.include_row_ids.is_some()
{
return Err(crate::Error::Unsupported {
message:
@@ -174,13 +197,7 @@ impl FullTextScan {
pub async fn scan(&self) -> crate::Result<FullTextScanPlan> {
CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?;
- let (partition_filter, row_filter) =
- bind_filters(&self.table, self.context.filter.as_ref())?;
- if 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 (partition_filter, _) = bind_filters(&self.table,
self.context.filter.as_ref())?;
let Some(snapshot) =
crate::table::time_travel::resolve_snapshot(&self.table).await? else {
return FullTextScanPlan::new(None, vec![]);
};
diff --git a/crates/paimon/src/table/global_index_search_filter.rs
b/crates/paimon/src/table/global_index_search_filter.rs
index 423bd867..28ac53ef 100644
--- a/crates/paimon/src/table/global_index_search_filter.rs
+++ b/crates/paimon/src/table/global_index_search_filter.rs
@@ -15,7 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-//! Exact row-ID filtering shared by data-evolution index searches.
+//! Exact row-ID filtering shared by global index searches.
use crate::spec::{Predicate, ROW_ID_FIELD_NAME};
use crate::table::{RowRange, Table};
diff --git a/docs/src/sql.md b/docs/src/sql.md
index b00cdfaf..0e2d1e2a 100644
--- a/docs/src/sql.md
+++ b/docs/src/sql.md
@@ -2238,9 +2238,20 @@ 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.
+Non-Data-Evolution tables also support row filters when they are append tables
+without primary keys, use Parquet, enable row tracking, and use full-text
`fast`
+mode. Scalar `fast`, `full`, and `detail` retain their coverage rules, with
data
+reads restricted to the full-text index coverage. Candidate refinement keeps
+the Rust default of `global-index.filter.refine-from-data=true`. Matching rows
+are checked against the pinned snapshot's live data and deletion vectors before
+full-text Top-K; filtering preserves the original index corpus and BM25 scores.
+Full-text `full`/`detail` row filtering on these tables remains unsupported.
+
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()`.
+accept plain text for the SQL table function. These pre-filters are configured
+through the Rust core builder's `with_filter()`; the SQL table function does
not
+push an outer `WHERE` into full-text search. Primary-key materialized reads use
+`execute_read()`.
## Time Travel
