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 000000000..cd810e20f --- /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 = 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::() + .unwrap() + .values() + .iter() + .copied() + }) + .collect::>(); + 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 eef7a1a91..bfca94b43 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 d37882857..c2a8c177b 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 3f7926b43..4aff6d42a 100644 --- a/crates/paimon/src/table/full_text_search_builder/scan.rs +++ b/crates/paimon/src/table/full_text_search_builder/scan.rs @@ -34,7 +34,8 @@ pub(super) struct PlanContext { impl PlanContext { pub(super) fn new(builder: &FullTextSearchBuilder<'_>) -> crate::Result { - 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() @@ -58,9 +59,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: @@ -193,12 +216,6 @@ impl FullTextScan { } 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); }; diff --git a/crates/paimon/src/table/global_index_search_filter.rs b/crates/paimon/src/table/global_index_search_filter.rs index 423bd8676..28ac53efb 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 b00cdfafe..0e2d1e2ab 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