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 316aa3e Support data evolution deletes with deletion vectors (#451)
316aa3e is described below
commit 316aa3ec410003fccb54a52925094147c09fc9d0
Author: Jingsong Lee <[email protected]>
AuthorDate: Sat Jul 4 23:03:51 2026 +0800
Support data evolution deletes with deletion vectors (#451)
---
crates/integrations/datafusion/src/delete.rs | 60 +++-
crates/integrations/datafusion/src/merge_into.rs | 37 +-
.../integrations/datafusion/tests/delete_tests.rs | 47 ++-
.../datafusion/tests/merge_into_tests.rs | 35 +-
crates/paimon/src/deletion_vector/core.rs | 40 +++
crates/paimon/src/deletion_vector/factory.rs | 2 +-
crates/paimon/src/lib.rs | 8 +-
crates/paimon/src/table/data_evolution_reader.rs | 239 +++++++++++--
crates/paimon/src/table/data_evolution_writer.rs | 382 ++++++++++++++++++++-
.../paimon/src/table/full_text_search_builder.rs | 34 +-
crates/paimon/src/table/global_index_scanner.rs | 158 ++++++++-
crates/paimon/src/table/mod.rs | 2 +-
crates/paimon/src/table/source.rs | 28 +-
crates/paimon/src/table/table_commit.rs | 111 +++++-
crates/paimon/src/table/vector_search_builder.rs | 48 ++-
crates/paimon/src/table/write_builder.rs | 7 +-
crates/paimon/src/tantivy/full_text_search.rs | 40 +++
crates/paimon/src/vector_search.rs | 40 +++
18 files changed, 1247 insertions(+), 71 deletions(-)
diff --git a/crates/integrations/datafusion/src/delete.rs
b/crates/integrations/datafusion/src/delete.rs
index c890cf6..832acd9 100644
--- a/crates/integrations/datafusion/src/delete.rs
+++ b/crates/integrations/datafusion/src/delete.rs
@@ -29,8 +29,8 @@ use paimon::table::{CopyOnWriteMergeWriter, Table};
use crate::error::to_datafusion_error;
use crate::merge_into::TempTableTracker;
use crate::merge_into::{
- build_partition_set_from_where, extract_tracking_columns,
is_delete_conflict, ok_result,
- register_cow_target_table, retry_on_conflict,
+ build_partition_set_from_where, extract_tracking_columns,
is_delete_conflict,
+ is_row_id_conflict, ok_result, register_cow_target_table,
retry_on_conflict,
};
use crate::sql_context::SQLContext;
@@ -60,9 +60,7 @@ pub(crate) async fn execute_delete(
let core_options = CoreOptions::new(schema.options());
if core_options.data_evolution_enabled() {
- return Err(DataFusionError::Plan(
- "DELETE on data-evolution tables is not yet supported".to_string(),
- ));
+ return execute_data_evolution_delete(ctx, delete, &table,
table_ref).await;
}
if !schema.trimmed_primary_keys().is_empty() {
return Err(DataFusionError::Plan(
@@ -73,6 +71,58 @@ pub(crate) async fn execute_delete(
execute_cow_delete(ctx, delete, &table, table_ref).await
}
+/// Execute DELETE on a data-evolution table with retry on row ID conflict.
+async fn execute_data_evolution_delete(
+ ctx: &SQLContext,
+ delete: &Delete,
+ table: &Table,
+ table_ref: &str,
+) -> DFResult<DataFrame> {
+ retry_on_conflict("Data-evolution DELETE", is_row_id_conflict, || {
+ execute_data_evolution_delete_once(ctx, delete, table, table_ref)
+ })
+ .await
+}
+
+/// Single attempt of data-evolution DELETE execution.
+async fn execute_data_evolution_delete_once(
+ ctx: &SQLContext,
+ delete: &Delete,
+ table: &Table,
+ table_ref: &str,
+) -> DFResult<DataFrame> {
+ let wb = table.new_write_builder();
+ let mut writer = wb.new_delete().map_err(to_datafusion_error)?;
+
+ let where_clause = match &delete.selection {
+ Some(expr) => format!(" WHERE {expr}"),
+ None => String::new(),
+ };
+ let query_sql = format!("SELECT \"_ROW_ID\" FROM
{table_ref}{where_clause}");
+ let batches = ctx.ctx().sql(&query_sql).await?.collect().await?;
+
+ let total_count: u64 = batches.iter().map(|batch| batch.num_rows() as
u64).sum();
+ if total_count == 0 {
+ return ok_result(ctx.ctx(), 0);
+ }
+
+ for batch in batches {
+ writer
+ .add_matched_batch(batch)
+ .map_err(to_datafusion_error)?;
+ }
+
+ let messages = writer.prepare_commit().await.map_err(to_datafusion_error)?;
+ if !messages.is_empty() {
+ wb.new_commit()
+ .commit(messages)
+ .await
+ .map_err(to_datafusion_error)?;
+ }
+
+ ok_result(ctx.ctx(), total_count)
+}
+
/// Execute DELETE on an append-only table with retry on delete conflict.
async fn execute_cow_delete(
ctx: &SQLContext,
diff --git a/crates/integrations/datafusion/src/merge_into.rs
b/crates/integrations/datafusion/src/merge_into.rs
index 79a8a0b..203538d 100644
--- a/crates/integrations/datafusion/src/merge_into.rs
+++ b/crates/integrations/datafusion/src/merge_into.rs
@@ -640,6 +640,11 @@ async fn execute_merge_into_once(
} else {
None
};
+ let delete_writer = if parsed.delete {
+ Some(wb.new_delete().map_err(to_datafusion_error)?)
+ } else {
+ None
+ };
let (target_ref, target_alias) = extract_table_ref(&merge.table)?;
let (source_ref, source_alias) = extract_source_ref(&merge.source)?;
@@ -685,7 +690,7 @@ async fn execute_merge_into_once(
// Separate matched and not-matched rows
let (matched_batches, not_matched_batches) =
split_by_row_id(&join_result)?;
- // 4. Handle matched rows (UPDATE)
+ // 4. Handle matched rows (UPDATE or DELETE)
if let Some(mut writer) = update_writer {
let upd = parsed.update.as_ref().unwrap();
let matched_count: usize = matched_batches.iter().map(|b|
b.num_rows()).sum();
@@ -702,6 +707,19 @@ async fn execute_merge_into_once(
total_count += matched_count as u64;
}
}
+ if let Some(mut writer) = delete_writer {
+ let matched_count: usize = matched_batches.iter().map(|b|
b.num_rows()).sum();
+ if matched_count > 0 {
+ for batch in &matched_batches {
+ writer
+ .add_matched_batch(batch.clone())
+ .map_err(to_datafusion_error)?;
+ }
+ let delete_messages =
writer.prepare_commit().await.map_err(to_datafusion_error)?;
+ all_messages.extend(delete_messages);
+ total_count += matched_count as u64;
+ }
+ }
// 5. Handle not-matched rows (INSERT)
if !parsed.inserts.is_empty() {
@@ -970,18 +988,20 @@ struct MergeUpdateClause {
/// Parsed merge clauses.
struct ParsedMergeClauses {
update: Option<MergeUpdateClause>,
+ delete: bool,
inserts: Vec<MergeInsertClause>,
}
/// Extract UPDATE and INSERT clauses from the MERGE AST.
fn extract_merge_clauses(merge: &Merge) -> DFResult<ParsedMergeClauses> {
let mut update: Option<MergeUpdateClause> = None;
+ let mut delete = false;
let mut inserts: Vec<MergeInsertClause> = Vec::new();
for clause in &merge.clauses {
match clause.clause_kind {
MergeClauseKind::Matched => {
- if update.is_some() {
+ if update.is_some() || delete {
return Err(DataFusionError::Plan(
"Multiple WHEN MATCHED clauses are not yet
supported".to_string(),
));
@@ -1020,10 +1040,7 @@ fn extract_merge_clauses(merge: &Merge) ->
DFResult<ParsedMergeClauses> {
update = Some(MergeUpdateClause { columns, exprs });
}
MergeAction::Delete { .. } => {
- return Err(DataFusionError::Plan(
- "WHEN MATCHED THEN DELETE is not supported for
data evolution tables"
- .to_string(),
- ));
+ delete = true;
}
MergeAction::Insert(_) => {
return Err(DataFusionError::Plan(
@@ -1077,13 +1094,17 @@ fn extract_merge_clauses(merge: &Merge) ->
DFResult<ParsedMergeClauses> {
}
}
- if update.is_none() && inserts.is_empty() {
+ if update.is_none() && !delete && inserts.is_empty() {
return Err(DataFusionError::Plan(
"MERGE INTO requires at least one WHEN MATCHED or WHEN NOT MATCHED
clause".to_string(),
));
}
- Ok(ParsedMergeClauses { update, inserts })
+ Ok(ParsedMergeClauses {
+ update,
+ delete,
+ inserts,
+ })
}
/// Extract table name and optional alias from a TableFactor.
diff --git a/crates/integrations/datafusion/tests/delete_tests.rs
b/crates/integrations/datafusion/tests/delete_tests.rs
index ac9ef95..aa247af 100644
--- a/crates/integrations/datafusion/tests/delete_tests.rs
+++ b/crates/integrations/datafusion/tests/delete_tests.rs
@@ -481,7 +481,7 @@ async fn test_delete_rejects_primary_key_table() {
}
#[tokio::test]
-async fn test_delete_rejects_data_evolution_table() {
+async fn test_delete_rejects_data_evolution_table_without_deletion_vectors() {
let (tmp, catalog) = create_test_env();
let sql_context = create_sql_context(catalog).await;
sql_context
@@ -503,12 +503,55 @@ async fn test_delete_rejects_data_evolution_table() {
assert_sql_error(
&sql_context,
"DELETE FROM paimon.test_db.de_t WHERE id = 1",
- "DELETE on data-evolution tables is not yet supported",
+ "deletion-vectors.enabled",
)
.await;
drop(tmp);
}
+#[tokio::test]
+async fn test_delete_data_evolution_table_with_deletion_vectors() {
+ let (tmp, catalog) = create_test_env();
+ let sql_context = create_sql_context(catalog).await;
+ sql_context
+ .sql("CREATE SCHEMA paimon.test_db")
+ .await
+ .unwrap();
+ sql_context
+ .sql(
+ "CREATE TABLE paimon.test_db.de_t (\
+ id INT NOT NULL, name VARCHAR, age INT\
+ ) WITH (\
+ 'row-tracking.enabled' = 'true',\
+ 'data-evolution.enabled' = 'true',\
+ 'deletion-vectors.enabled' = 'true'\
+ )",
+ )
+ .await
+ .unwrap();
+ exec(
+ &sql_context,
+ "INSERT INTO paimon.test_db.de_t (id, name, age) VALUES (1, 'a', 10),
(2, 'b', 20), (3, 'c', 30)",
+ )
+ .await;
+
+ let cnt = dml_count(&sql_context, "DELETE FROM paimon.test_db.de_t WHERE
id = 2").await;
+ assert_eq!(cnt, 1);
+
+ assert_eq!(
+ query_int_str_int(
+ &sql_context,
+ "SELECT id, name, age FROM paimon.test_db.de_t ORDER BY id",
+ )
+ .await,
+ vec![(1, "a".into(), 10), (3, "c".into(), 30)]
+ );
+
+ let second_cnt = dml_count(&sql_context, "DELETE FROM paimon.test_db.de_t
WHERE id = 2").await;
+ assert_eq!(second_cnt, 0);
+ drop(tmp);
+}
+
#[tokio::test]
async fn test_delete_rejects_table_alias() {
let (_tmp, sql_context) = setup().await;
diff --git a/crates/integrations/datafusion/tests/merge_into_tests.rs
b/crates/integrations/datafusion/tests/merge_into_tests.rs
index c6271ed..f3e9358 100644
--- a/crates/integrations/datafusion/tests/merge_into_tests.rs
+++ b/crates/integrations/datafusion/tests/merge_into_tests.rs
@@ -827,20 +827,31 @@ async fn
test_merge_into_rejects_duplicate_matched_updates() {
}
#[tokio::test]
-async fn test_rejects_when_matched_delete() {
+async fn test_when_matched_delete_with_deletion_vectors() {
let (_tmp, catalog) = create_test_env();
let sql_context = create_sql_context(catalog.clone()).await;
setup_data_evolution_table(&sql_context).await;
sql_context
- .sql("INSERT INTO paimon.test_db.target (id, name, value) VALUES (1,
'alice', 10)")
+ .sql("INSERT INTO paimon.test_db.target (id, name, value) VALUES (1,
'alice', 10), (2, 'bob', 20)")
.await
.unwrap()
.collect()
.await
.unwrap();
- enable_data_evolution(&sql_context).await;
+ sql_context
+ .sql(
+ "ALTER TABLE paimon.test_db.target SET TBLPROPERTIES(\
+ 'data-evolution.enabled' = 'true',\
+ 'deletion-vectors.enabled' = 'true'\
+ )",
+ )
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
register_source(
&sql_context,
@@ -848,13 +859,23 @@ async fn test_rejects_when_matched_delete() {
)
.await;
- assert_merge_error(
+ sql_context
+ .sql(
+ "MERGE INTO paimon.test_db.target t USING paimon.test_db.src_del s
ON t.id = s.id \
+ WHEN MATCHED THEN DELETE",
+ )
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+
+ let rows = collect_rows_3col(
&sql_context,
- "MERGE INTO paimon.test_db.target t USING paimon.test_db.src_del s ON
t.id = s.id \
- WHEN MATCHED THEN DELETE",
- "WHEN MATCHED THEN DELETE is not supported",
+ "SELECT id, name, value FROM paimon.test_db.target ORDER BY id",
)
.await;
+ assert_eq!(rows, vec![(2, "bob".to_string(), 20)]);
}
#[tokio::test]
diff --git a/crates/paimon/src/deletion_vector/core.rs
b/crates/paimon/src/deletion_vector/core.rs
index 831f1d0..11c2d06 100644
--- a/crates/paimon/src/deletion_vector/core.rs
+++ b/crates/paimon/src/deletion_vector/core.rs
@@ -49,6 +49,16 @@ impl DeletionVector {
}
}
+ /// Clone the underlying bitmap for mutation by writers.
+ pub(crate) fn to_bitmap(&self) -> RoaringBitmap {
+ (*self.bitmap).clone()
+ }
+
+ /// Number of deleted positions in this vector.
+ pub fn cardinality(&self) -> u64 {
+ self.bitmap.len()
+ }
+
/// Returns an iterator over deleted positions that supports
[DeletionVectorIterator::advance_to].
/// Required for efficient row selection building when skipping row groups
(avoid re-scanning
/// deletes in skipped ranges).
@@ -66,6 +76,36 @@ impl DeletionVector {
self.bitmap.is_empty()
}
+ /// Serialize using Java `BitmapDeletionVector` format:
+ /// `i32 bitmapLength | i32 magic | roaring bitmap bytes | i32 crc`.
+ pub(crate) fn serialize_to_bytes(&self) -> crate::Result<Vec<u8>> {
+ let mut bitmap_bytes = Vec::new();
+ self.bitmap
+ .serialize_into(&mut bitmap_bytes)
+ .map_err(|e| crate::Error::DataInvalid {
+ message: format!("Failed to serialize RoaringBitmap: {e}"),
+ source: Some(Box::new(e)),
+ })?;
+
+ let bitmap_length =
+ i32::try_from(MAGIC_NUMBER_SIZE_BYTES +
bitmap_bytes.len()).map_err(|_| {
+ crate::Error::DataInvalid {
+ message: "Deletion vector bitmap is too large to
serialize".to_string(),
+ source: None,
+ }
+ })?;
+
+ let mut payload = Vec::with_capacity(8 + bitmap_bytes.len() + 4);
+ payload.extend_from_slice(&bitmap_length.to_be_bytes());
+ payload.extend_from_slice(&(MAGIC_NUMBER as i32).to_be_bytes());
+ payload.extend_from_slice(&bitmap_bytes);
+
+ let mut crc = crc32fast::Hasher::new();
+ crc.update(&payload[4..]);
+ payload.extend_from_slice(&(crc.finalize() as i32).to_be_bytes());
+ Ok(payload)
+ }
+
/// Get the underlying bitmap (read-only)
#[cfg(test)]
fn bitmap(&self) -> &RoaringBitmap {
diff --git a/crates/paimon/src/deletion_vector/factory.rs
b/crates/paimon/src/deletion_vector/factory.rs
index 927d92f..cb5a47e 100644
--- a/crates/paimon/src/deletion_vector/factory.rs
+++ b/crates/paimon/src/deletion_vector/factory.rs
@@ -63,7 +63,7 @@ impl DeletionVectorFactory {
/// Read a single DeletionVector from storage using DeletionFile
(path/offset/length).
/// Same as Java's DeletionVector.read(FileIO, DeletionFile).
- async fn read(file_io: &FileIO, df: &crate::DeletionFile) ->
Result<DeletionVector> {
+ pub(crate) async fn read(file_io: &FileIO, df: &crate::DeletionFile) ->
Result<DeletionVector> {
let input = file_io.new_input(df.path())?;
let reader = input.reader().await?;
let offset = df.offset() as u64;
diff --git a/crates/paimon/src/lib.rs b/crates/paimon/src/lib.rs
index a697218..8c26fe6 100644
--- a/crates/paimon/src/lib.rs
+++ b/crates/paimon/src/lib.rs
@@ -45,8 +45,8 @@ pub use catalog::CatalogFactory;
pub use catalog::FileSystemCatalog;
pub use table::{
- CommitMessage, DataEvolutionWriter, DataSplit, DataSplitBuilder,
DeletionFile, PartitionBucket,
- Plan, RESTEnv, RESTSnapshotCommit, ReadBuilder, RenamingSnapshotCommit,
RowRange, ScanTrace,
- SnapshotCommit, SnapshotManager, Table, TableCommit, TableRead, TableScan,
TableUpdate,
- TableWrite, TagManager, WriteBuilder,
+ CommitMessage, DataEvolutionDeleteWriter, DataEvolutionWriter, DataSplit,
DataSplitBuilder,
+ DeletionFile, PartitionBucket, Plan, RESTEnv, RESTSnapshotCommit,
ReadBuilder,
+ RenamingSnapshotCommit, RowRange, ScanTrace, SnapshotCommit,
SnapshotManager, Table,
+ TableCommit, TableRead, TableScan, TableUpdate, TableWrite, TagManager,
WriteBuilder,
};
diff --git a/crates/paimon/src/table/data_evolution_reader.rs
b/crates/paimon/src/table/data_evolution_reader.rs
index 8f4a5cb..e016240 100644
--- a/crates/paimon/src/table/data_evolution_reader.rs
+++ b/crates/paimon/src/table/data_evolution_reader.rs
@@ -20,6 +20,7 @@ use super::data_file_reader::{
DataFileReader,
};
use crate::arrow::build_target_arrow_schema;
+use crate::deletion_vector::{DeletionVector, DeletionVectorFactory};
use crate::io::FileIO;
use crate::spec::{DataField, DataFileMeta, DataType, ROW_ID_FIELD_NAME};
use crate::table::blob_file_writer::is_blob_file_name;
@@ -30,6 +31,7 @@ use crate::{DataSplit, Error};
use arrow_array::{Array, Int64Array, RecordBatch};
use async_stream::try_stream;
use futures::StreamExt;
+use roaring::RoaringBitmap;
use std::collections::{HashMap, HashSet};
use std::sync::Arc;
@@ -140,6 +142,12 @@ impl DataEvolutionReader {
if is_raw_convertible(split.data_files()) {
for file_meta in split.data_files().to_vec() {
+ let deletion_vector = read_file_deletion_vector(
+ &self.file_io,
+ &split,
+ &file_meta,
+ )
+ .await?;
let data_fields: Option<Vec<DataField>> =
if file_meta.schema_id != self.table_schema_id {
let data_schema =
@@ -153,11 +161,17 @@ impl DataEvolutionReader {
let effective_row_ranges = if has_row_id {
row_ranges.clone() } else { None };
let selected_row_ids = if self.row_id_index.is_some()
&& has_row_id {
- effective_row_ranges.as_ref().map(|ranges| {
+ selected_absolute_row_ranges_for_file(
+ file_meta.first_row_id.unwrap(),
+ file_meta.row_count,
+ effective_row_ranges.as_deref(),
+ deletion_vector.as_deref(),
+ )?
+ .map(|ranges| {
expand_selected_row_ids(
file_meta.first_row_id.unwrap(),
file_meta.row_count,
- ranges,
+ &ranges,
)
})
} else {
@@ -171,7 +185,7 @@ impl DataEvolutionReader {
&split,
file_meta,
data_fields,
- None,
+ deletion_vector,
effective_row_ranges,
)?;
while let Some(batch) = stream.next().await {
@@ -200,15 +214,28 @@ impl DataEvolutionReader {
}
} else {
let prepared_group =
PreparedMergeGroup::new(split.data_files())?;
+ let anchor_deletion_vector = read_anchor_deletion_vector(
+ &self.file_io,
+ &split,
+ &prepared_group.files,
+ )
+ .await?;
let effective_row_ranges = row_ranges.clone();
- let expected_output_rows = count_selected_rows(
+ let selected_ranges =
selected_absolute_row_ranges_for_file(
prepared_group.first_row_id,
prepared_group.logical_row_count,
effective_row_ranges.as_deref(),
+ anchor_deletion_vector
+ .as_ref()
+ .map(|ctx| ctx.deletion_vector.as_ref()),
)?;
+ let expected_output_rows = match selected_ranges.as_ref() {
+ Some(ranges) => ranges.iter().map(|r| r.count() as
usize).sum(),
+ None => prepared_group.logical_row_count as usize,
+ };
let selected_row_ids = if self.row_id_index.is_some() {
- effective_row_ranges.as_ref().map(|ranges| {
+ selected_ranges.as_ref().map(|ranges| {
expand_selected_row_ids(
prepared_group.first_row_id,
prepared_group.logical_row_count,
@@ -226,6 +253,7 @@ impl DataEvolutionReader {
&prepared_group,
effective_row_ranges,
expected_output_rows,
+ anchor_deletion_vector,
)?;
while let Some(batch) = merge_stream.next().await {
let batch = batch?;
@@ -260,6 +288,7 @@ impl DataEvolutionReader {
prepared_group: &PreparedMergeGroup,
row_ranges: Option<Vec<RowRange>>,
expected_output_rows: usize,
+ anchor_deletion_vector: Option<DeletionVectorContext>,
) -> crate::Result<ArrowRecordBatchStream> {
if prepared_group.files.is_empty() {
return Ok(futures::stream::empty().boxed());
@@ -274,6 +303,7 @@ impl DataEvolutionReader {
let table_fields = self.table_fields.clone();
let blob_descriptor_fields = self.blob_descriptor_fields.clone();
let blob_as_descriptor = self.blob_as_descriptor;
+ let anchor_deletion_vector = anchor_deletion_vector.clone();
// Batch size for column-merge output. Matches the default Parquet
reader batch size.
const MERGE_BATCH_SIZE: usize = 1024;
let target_schema = build_target_arrow_schema(&read_type)?;
@@ -340,6 +370,7 @@ impl DataEvolutionReader {
table_schema_id,
table_fields.clone(),
blob_as_descriptor,
+ anchor_deletion_vector.as_ref(),
)
.map(Some)
}
@@ -501,6 +532,7 @@ fn open_source_stream(
table_schema_id: i64,
table_fields: Vec<DataField>,
blob_as_descriptor: bool,
+ anchor_deletion_vector: Option<&DeletionVectorContext>,
) -> crate::Result<ArrowRecordBatchStream> {
let file_reader = DataFileReader::new(
file_io,
@@ -515,13 +547,16 @@ fn open_source_stream(
match source {
FieldSource::DataFile {
file, data_fields, ..
- } => file_reader.read_single_file_stream(
- split,
- file.as_ref().clone(),
- data_fields.clone(),
- None,
- row_ranges,
- ),
+ } => {
+ let deletion_vector = shifted_deletion_vector_for_file(file,
anchor_deletion_vector)?;
+ file_reader.read_single_file_stream(
+ split,
+ file.as_ref().clone(),
+ data_fields.clone(),
+ deletion_vector,
+ row_ranges,
+ )
+ }
FieldSource::BlobBunch {
bunch, data_fields, ..
} => read_bunch_files_stream(
@@ -530,6 +565,7 @@ fn open_source_stream(
bunch.files.clone(),
data_fields.clone(),
row_ranges,
+ anchor_deletion_vector.cloned(),
),
FieldSource::VectorBunch {
bunch, data_fields, ..
@@ -539,6 +575,7 @@ fn open_source_stream(
bunch.files.clone(),
data_fields.clone(),
row_ranges,
+ anchor_deletion_vector.cloned(),
),
}
}
@@ -549,15 +586,18 @@ fn read_bunch_files_stream(
files: Vec<DataFileMeta>,
data_fields: Option<Vec<DataField>>,
row_ranges: Option<Vec<RowRange>>,
+ anchor_deletion_vector: Option<DeletionVectorContext>,
) -> crate::Result<ArrowRecordBatchStream> {
let split = split.clone();
Ok(try_stream! {
for file in files {
+ let deletion_vector =
+ shifted_deletion_vector_for_file(&file,
anchor_deletion_vector.as_ref())?;
let mut stream = file_reader.read_single_file_stream(
&split,
file,
data_fields.clone(),
- None,
+ deletion_vector,
row_ranges.clone(),
)?;
while let Some(batch) = stream.next().await {
@@ -568,6 +608,165 @@ fn read_bunch_files_stream(
.boxed())
}
+#[derive(Debug, Clone)]
+struct DeletionVectorContext {
+ first_row_id: i64,
+ deletion_vector: Arc<DeletionVector>,
+}
+
+async fn read_file_deletion_vector(
+ file_io: &FileIO,
+ split: &DataSplit,
+ file: &DataFileMeta,
+) -> crate::Result<Option<Arc<DeletionVector>>> {
+ let Some(deletion_file) = split.deletion_file_for_data_file(file) else {
+ return Ok(None);
+ };
+ Ok(Some(Arc::new(
+ DeletionVectorFactory::read(file_io, deletion_file).await?,
+ )))
+}
+
+async fn read_anchor_deletion_vector(
+ file_io: &FileIO,
+ split: &DataSplit,
+ files: &[DataFileMeta],
+) -> crate::Result<Option<DeletionVectorContext>> {
+ let anchor = crate::table::source::data_evolution_anchor_file(files)?;
+ let Some(deletion_file) = split.deletion_file_for_data_file(anchor) else {
+ return Ok(None);
+ };
+ let first_row_id = anchor.first_row_id.ok_or_else(|| Error::DataInvalid {
+ message: format!(
+ "Data-evolution anchor file '{}' is missing first_row_id",
+ anchor.file_name
+ ),
+ source: None,
+ })?;
+ Ok(Some(DeletionVectorContext {
+ first_row_id,
+ deletion_vector: Arc::new(DeletionVectorFactory::read(file_io,
deletion_file).await?),
+ }))
+}
+
+fn shifted_deletion_vector_for_file(
+ file: &DataFileMeta,
+ context: Option<&DeletionVectorContext>,
+) -> crate::Result<Option<Arc<DeletionVector>>> {
+ let Some(context) = context else {
+ return Ok(None);
+ };
+ let Some(file_first_row_id) = file.first_row_id else {
+ return Ok(None);
+ };
+
+ if file_first_row_id == context.first_row_id {
+ return Ok(Some(context.deletion_vector.clone()));
+ }
+
+ let file_end = file_first_row_id + file.row_count - 1;
+ let mut bitmap = RoaringBitmap::new();
+ for deleted in context.deletion_vector.iter() {
+ let row_id = context.first_row_id + deleted as i64;
+ if row_id < file_first_row_id || row_id > file_end {
+ continue;
+ }
+ let local = u32::try_from(row_id - file_first_row_id).map_err(|_|
Error::DataInvalid {
+ message: format!(
+ "Deleted row id {row_id} cannot be represented as a local
deletion-vector position for file '{}'",
+ file.file_name
+ ),
+ source: None,
+ })?;
+ bitmap.insert(local);
+ }
+
+ if bitmap.is_empty() {
+ Ok(None)
+ } else {
+ Ok(Some(Arc::new(DeletionVector::from_bitmap(bitmap))))
+ }
+}
+
+fn selected_absolute_row_ranges_for_file(
+ first_row_id: i64,
+ row_count: i64,
+ row_ranges: Option<&[RowRange]>,
+ deletion_vector: Option<&DeletionVector>,
+) -> crate::Result<Option<Vec<RowRange>>> {
+ let has_ranges = row_ranges.is_some();
+ let has_deletion_vector = deletion_vector.is_some_and(|dv| !dv.is_empty());
+ if !has_ranges && !has_deletion_vector {
+ return Ok(None);
+ }
+ if row_count == 0 {
+ return Ok(Some(Vec::new()));
+ }
+
+ let mut local_ranges = if let Some(dv) = deletion_vector {
+ non_deleted_local_ranges(row_count, dv)
+ } else {
+ vec![RowRange::new(0, row_count - 1)]
+ };
+
+ if let Some(ranges) = row_ranges {
+ let selected = ranges
+ .iter()
+ .filter_map(|range| {
+ range
+ .intersect_inclusive(first_row_id, first_row_id +
row_count - 1)
+ .map(|range| {
+ RowRange::new(range.from() - first_row_id, range.to()
- first_row_id)
+ })
+ })
+ .collect::<Vec<_>>();
+ local_ranges = intersect_local_ranges(&local_ranges, &selected);
+ }
+
+ let absolute = local_ranges
+ .into_iter()
+ .map(|range| RowRange::new(first_row_id + range.from(), first_row_id +
range.to()))
+ .collect::<Vec<_>>();
+ Ok(Some(absolute))
+}
+
+fn non_deleted_local_ranges(row_count: i64, deletion_vector: &DeletionVector)
-> Vec<RowRange> {
+ let mut ranges = Vec::new();
+ let mut cursor = 0i64;
+ for deleted in deletion_vector.iter() {
+ let deleted = deleted as i64;
+ if deleted >= row_count {
+ break;
+ }
+ if deleted > cursor {
+ ranges.push(RowRange::new(cursor, deleted - 1));
+ }
+ cursor = deleted + 1;
+ }
+ if cursor < row_count {
+ ranges.push(RowRange::new(cursor, row_count - 1));
+ }
+ ranges
+}
+
+fn intersect_local_ranges(left: &[RowRange], right: &[RowRange]) ->
Vec<RowRange> {
+ let mut result = Vec::new();
+ let (mut i, mut j) = (0usize, 0usize);
+ while i < left.len() && j < right.len() {
+ let from = left[i].from().max(right[j].from());
+ let to = left[i].to().min(right[j].to());
+ if from <= to {
+ result.push(RowRange::new(from, to));
+ }
+ if left[i].to() < right[j].to() {
+ i += 1;
+ } else {
+ j += 1;
+ }
+ }
+ result
+}
+
#[derive(Debug, Clone)]
struct PreparedMergeGroup {
files: Vec<DataFileMeta>,
@@ -1317,20 +1516,6 @@ fn normalize_merge_group(files: Vec<DataFileMeta>) ->
crate::Result<Vec<DataFile
Ok(out)
}
-fn count_selected_rows(
- first_row_id: i64,
- row_count: i64,
- row_ranges: Option<&[RowRange]>,
-) -> crate::Result<usize> {
- match row_ranges {
- Some(ranges) => Ok(expand_selected_row_ids(first_row_id, row_count,
ranges).len()),
- None => usize::try_from(row_count).map_err(|e| Error::DataInvalid {
- message: format!("Invalid logical row count {row_count}"),
- source: Some(Box::new(e)),
- }),
- }
-}
-
#[cfg(test)]
mod tests {
use super::*;
diff --git a/crates/paimon/src/table/data_evolution_writer.rs
b/crates/paimon/src/table/data_evolution_writer.rs
index 3e12316..2c51f92 100644
--- a/crates/paimon/src/table/data_evolution_writer.rs
+++ b/crates/paimon/src/table/data_evolution_writer.rs
@@ -26,19 +26,34 @@
//! This separation allows callers to compose multiple operations into a
single commit,
//! similar to Iceberg's Transaction/Action pattern.
+use crate::deletion_vector::{DeletionVector, DeletionVectorFactory};
use crate::io::FileIO;
-use crate::spec::{BinaryRow, CoreOptions, DataField, DataFileMeta,
PartitionComputer};
+use crate::spec::{
+ BinaryRow, CoreOptions, DataField, DataFileMeta, DeletionVectorMeta,
FileKind, IndexFileMeta,
+ IndexManifest, PartitionComputer,
+};
use crate::table::commit_message::CommitMessage;
use crate::table::data_file_writer::DataFileWriter;
+use crate::table::source::data_evolution_anchor_file;
use crate::table::stats_filter::group_by_overlapping_row_id;
use crate::table::DataSplitBuilder;
+use crate::table::SnapshotManager;
use crate::table::Table;
use crate::Result;
use arrow_array::{Array, ArrayRef, Int64Array, RecordBatch};
use arrow_select::concat::concat_batches;
use arrow_select::interleave::interleave;
+use bytes::Bytes;
use futures::TryStreamExt;
+use indexmap::IndexMap;
+use roaring::RoaringBitmap;
use std::collections::{HashMap, HashSet};
+use uuid::Uuid;
+
+const DELETION_VECTORS_INDEX_TYPE: &str = "DELETION_VECTORS";
+const DELETION_VECTORS_INDEX_VERSION_V1: u8 = 1;
+const INDEX_DIR: &str = "index";
+const MANIFEST_DIR: &str = "manifest";
/// Engine-agnostic writer for partial-column updates via `_ROW_ID`.
///
@@ -369,6 +384,371 @@ impl DataEvolutionWriter {
}
}
+/// Engine-agnostic DELETE writer for data evolution tables.
+///
+/// DELETE is represented by a deletion-vector index file keyed by the normal
+/// anchor file of each data-evolution row-id group. The data files themselves
+/// are not rewritten.
+#[must_use = "writer must be used to call prepare_commit()"]
+pub struct DataEvolutionDeleteWriter {
+ table: Table,
+ row_ids: Vec<i64>,
+}
+
+impl DataEvolutionDeleteWriter {
+ pub fn new(table: &Table) -> Result<Self> {
+ let schema = table.schema();
+ let core_options = CoreOptions::new(schema.options());
+
+ if !core_options.data_evolution_enabled() {
+ return Err(crate::Error::Unsupported {
+ message:
+ "DELETE is only supported for tables with
'data-evolution.enabled' = 'true'"
+ .to_string(),
+ });
+ }
+ if !core_options.row_tracking_enabled() {
+ return Err(crate::Error::Unsupported {
+ message: "DELETE requires 'row-tracking.enabled' =
'true'".to_string(),
+ });
+ }
+ if !core_options.deletion_vectors_enabled() {
+ return Err(crate::Error::Unsupported {
+ message:
+ "DELETE on data evolution tables requires
'deletion-vectors.enabled' = 'true'"
+ .to_string(),
+ });
+ }
+ if !schema.trimmed_primary_keys().is_empty() {
+ return Err(crate::Error::Unsupported {
+ message: "DELETE on data evolution tables does not support
primary keys"
+ .to_string(),
+ });
+ }
+
+ Ok(Self {
+ table: table.clone(),
+ row_ids: Vec::new(),
+ })
+ }
+
+ pub fn add_row_ids<I>(&mut self, row_ids: I) -> Result<()>
+ where
+ I: IntoIterator<Item = i64>,
+ {
+ self.row_ids.extend(row_ids);
+ Ok(())
+ }
+
+ pub fn add_matched_batch(&mut self, batch: RecordBatch) -> Result<()> {
+ if batch.num_rows() == 0 {
+ return Ok(());
+ }
+
+ let row_id_col = row_id_column(&batch)?;
+ validate_row_id_not_null(row_id_col)?;
+ self.row_ids
+ .extend((0..row_id_col.len()).map(|row_idx|
row_id_col.value(row_idx)));
+ Ok(())
+ }
+
+ #[must_use = "commit messages must be passed to TableCommit"]
+ pub async fn prepare_commit(mut self) -> Result<Vec<CommitMessage>> {
+ dedup_i64_in_place(&mut self.row_ids);
+ if self.row_ids.is_empty() {
+ return Ok(Vec::new());
+ }
+
+ let scan = self
+ .table
+ .new_read_builder()
+ .new_scan()
+ .with_scan_all_files();
+ let plan = scan.plan().await?;
+ let mut file_index = Vec::new();
+
+ for split in plan.splits() {
+ let partition = split.partition().to_serialized_bytes();
+ let bucket = split.bucket();
+ let snapshot_id = split.snapshot_id();
+ let files = split
+ .data_files()
+ .iter()
+ .filter(|file| file.first_row_id.is_some())
+ .cloned()
+ .collect::<Vec<_>>();
+
+ for group in group_by_overlapping_row_id(files) {
+ let anchor = data_evolution_anchor_file(&group)?;
+ let first_row_id =
+ anchor
+ .first_row_id
+ .ok_or_else(|| crate::Error::DataInvalid {
+ message: format!(
+ "Data-evolution anchor file '{}' is missing
first_row_id",
+ anchor.file_name
+ ),
+ source: None,
+ })?;
+ let last_row_id = first_row_id + anchor.row_count - 1;
+ file_index.push(DeleteFileRowRange {
+ first_row_id,
+ last_row_id,
+ partition: partition.clone(),
+ bucket,
+ snapshot_id,
+ anchor_file_name: anchor.file_name.clone(),
+ });
+ }
+ }
+ file_index.sort_by_key(|range| range.first_row_id);
+
+ if file_index.is_empty() {
+ return Err(crate::Error::DataInvalid {
+ message: "No files with row tracking found in target
table".to_string(),
+ source: None,
+ });
+ }
+
+ let mut deletes_by_bucket: HashMap<(Vec<u8>, i32), BucketDeletePlan> =
HashMap::new();
+ for row_id in &self.row_ids {
+ let (file_pos, file_range) =
+ find_delete_owning_file(&file_index, *row_id).ok_or_else(|| {
+ crate::Error::DataInvalid {
+ message: format!("No file found for _ROW_ID {row_id}"),
+ source: None,
+ }
+ })?;
+ let local_position = u32::try_from(row_id -
file_range.first_row_id).map_err(|_| {
+ crate::Error::DataInvalid {
+ message: format!(
+ "_ROW_ID {row_id} is too large to encode in a deletion
vector"
+ ),
+ source: None,
+ }
+ })?;
+
+ let key = (file_range.partition.clone(), file_range.bucket);
+ let entry = deletes_by_bucket
+ .entry(key)
+ .or_insert_with(|| BucketDeletePlan {
+ check_from_snapshot: file_range.snapshot_id,
+ deletes_by_anchor: HashMap::new(),
+ });
+ entry.check_from_snapshot =
entry.check_from_snapshot.min(file_range.snapshot_id);
+ entry
+ .deletes_by_anchor
+ .entry(file_index[file_pos].anchor_file_name.clone())
+ .or_default()
+ .insert(local_position);
+ }
+
+ let mut messages = Vec::new();
+ for ((partition, bucket), delete_plan) in deletes_by_bucket {
+ if let Some(message) = self
+ .prepare_bucket_delete_message(partition, bucket, delete_plan)
+ .await?
+ {
+ messages.push(message);
+ }
+ }
+
+ Ok(messages)
+ }
+
+ async fn prepare_bucket_delete_message(
+ &self,
+ partition: Vec<u8>,
+ bucket: i32,
+ delete_plan: BucketDeletePlan,
+ ) -> Result<Option<CommitMessage>> {
+ let (mut bitmaps, deleted_index_files) = self
+ .read_existing_bucket_deletion_vectors(
+ &partition,
+ bucket,
+ delete_plan.check_from_snapshot,
+ )
+ .await?;
+ let mut changed = false;
+
+ for (anchor_file_name, positions) in delete_plan.deletes_by_anchor {
+ let bitmap = bitmaps.entry(anchor_file_name).or_default();
+ for position in positions {
+ changed |= bitmap.insert(position);
+ }
+ }
+
+ if !changed {
+ return Ok(None);
+ }
+
+ let new_index_file =
self.write_deletion_vector_index_file(bitmaps).await?;
+ let mut message = CommitMessage::new(partition, bucket, vec![]);
+ message.check_from_snapshot = Some(delete_plan.check_from_snapshot);
+ message.new_index_files = vec![new_index_file];
+ message.deleted_index_files = deleted_index_files;
+ Ok(Some(message))
+ }
+
+ async fn read_existing_bucket_deletion_vectors(
+ &self,
+ partition: &[u8],
+ bucket: i32,
+ snapshot_id: i64,
+ ) -> Result<(IndexMap<String, RoaringBitmap>, Vec<IndexFileMeta>)> {
+ let snapshot_manager = SnapshotManager::new(
+ self.table.file_io().clone(),
+ self.table.location().to_string(),
+ );
+ let snapshot = snapshot_manager.get_snapshot(snapshot_id).await?;
+ let Some(index_manifest_name) = snapshot.index_manifest() else {
+ return Ok((IndexMap::new(), Vec::new()));
+ };
+
+ let manifest_path = format!(
+ "{}/{MANIFEST_DIR}/{}",
+ self.table.location().trim_end_matches('/'),
+ index_manifest_name
+ );
+ let index_entries = IndexManifest::read(self.table.file_io(),
&manifest_path).await?;
+ let mut bitmaps = IndexMap::new();
+ let mut deleted_index_files = Vec::new();
+
+ for entry in index_entries {
+ if entry.kind != FileKind::Add
+ || entry.bucket != bucket
+ || entry.partition != partition
+ || entry.index_file.index_type != DELETION_VECTORS_INDEX_TYPE
+ {
+ continue;
+ }
+ deleted_index_files.push(entry.index_file.clone());
+ let Some(ranges) =
entry.index_file.deletion_vectors_ranges.as_ref() else {
+ continue;
+ };
+ let index_path = format!(
+ "{}/{INDEX_DIR}/{}",
+ self.table.location().trim_end_matches('/'),
+ entry.index_file.file_name
+ );
+ for (data_file_name, meta) in ranges {
+ let deletion_file = crate::DeletionFile::new(
+ index_path.clone(),
+ meta.offset as i64,
+ meta.length as i64,
+ meta.cardinality,
+ );
+ let bitmap = DeletionVectorFactory::read(self.table.file_io(),
&deletion_file)
+ .await?
+ .to_bitmap();
+ bitmaps.insert(data_file_name.clone(), bitmap);
+ }
+ }
+
+ Ok((bitmaps, deleted_index_files))
+ }
+
+ async fn write_deletion_vector_index_file(
+ &self,
+ mut bitmaps: IndexMap<String, RoaringBitmap>,
+ ) -> Result<IndexFileMeta> {
+ bitmaps.sort_keys();
+
+ let file_name = format!("index-{}-1", Uuid::new_v4());
+ let table_path = self.table.location().trim_end_matches('/');
+ let index_dir = format!("{table_path}/{INDEX_DIR}");
+ self.table.file_io().mkdirs(&index_dir).await?;
+ let path = format!("{index_dir}/{file_name}");
+
+ let mut bytes = vec![DELETION_VECTORS_INDEX_VERSION_V1];
+ let mut ranges = IndexMap::new();
+ for (data_file_name, bitmap) in bitmaps {
+ if bitmap.is_empty() {
+ continue;
+ }
+ let offset = i32::try_from(bytes.len()).map_err(|_|
crate::Error::DataInvalid {
+ message: "Deletion-vector index file is too large".to_string(),
+ source: None,
+ })?;
+ let deletion_vector = DeletionVector::from_bitmap(bitmap);
+ let serialized = deletion_vector.serialize_to_bytes()?;
+ let length = i32::from_be_bytes(
+ serialized[0..4]
+ .try_into()
+ .expect("serialized deletion vector has length prefix"),
+ );
+ ranges.insert(
+ data_file_name,
+ DeletionVectorMeta {
+ offset,
+ length,
+ cardinality: Some(deletion_vector.cardinality() as i64),
+ },
+ );
+ bytes.extend_from_slice(&serialized);
+ }
+
+ let file_size = i32::try_from(bytes.len()).map_err(|_|
crate::Error::DataInvalid {
+ message: "Deletion-vector index file is too large".to_string(),
+ source: None,
+ })?;
+ let row_count = i32::try_from(ranges.len()).map_err(|_|
crate::Error::DataInvalid {
+ message: "Deletion-vector index file has too many
entries".to_string(),
+ source: None,
+ })?;
+ self.table
+ .file_io()
+ .new_output(&path)?
+ .write(Bytes::from(bytes))
+ .await?;
+
+ Ok(IndexFileMeta {
+ index_type: DELETION_VECTORS_INDEX_TYPE.to_string(),
+ file_name,
+ file_size,
+ row_count,
+ deletion_vectors_ranges: Some(ranges),
+ global_index_meta: None,
+ })
+ }
+}
+
+struct DeleteFileRowRange {
+ first_row_id: i64,
+ last_row_id: i64,
+ partition: Vec<u8>,
+ bucket: i32,
+ snapshot_id: i64,
+ anchor_file_name: String,
+}
+
+struct BucketDeletePlan {
+ check_from_snapshot: i64,
+ deletes_by_anchor: HashMap<String, HashSet<u32>>,
+}
+
+fn find_delete_owning_file(
+ file_index: &[DeleteFileRowRange],
+ row_id: i64,
+) -> Option<(usize, &DeleteFileRowRange)> {
+ let pos = file_index.partition_point(|f| f.first_row_id <= row_id);
+ if pos == 0 {
+ return None;
+ }
+ let idx = pos - 1;
+ let candidate = &file_index[idx];
+ if row_id <= candidate.last_row_id {
+ Some((idx, candidate))
+ } else {
+ None
+ }
+}
+
+fn dedup_i64_in_place(values: &mut Vec<i64>) {
+ let mut seen = HashSet::new();
+ values.retain(|value| seen.insert(*value));
+}
+
/// Binary search for the file that owns a given row_id.
fn find_owning_file(file_index: &[FileRowRange], row_id: i64) ->
Option<(usize, &FileRowRange)> {
let pos = file_index.partition_point(|f| f.first_row_id <= row_id);
diff --git a/crates/paimon/src/table/full_text_search_builder.rs
b/crates/paimon/src/table/full_text_search_builder.rs
index 04ffe21..b24e82e 100644
--- a/crates/paimon/src/table/full_text_search_builder.rs
+++ b/crates/paimon/src/table/full_text_search_builder.rs
@@ -24,7 +24,10 @@ use crate::spec::{
CoreOptions, DataField, FileKind, GlobalIndexSearchMode, IndexFileMeta,
IndexManifest,
IndexManifestEntry, ROW_ID_FIELD_NAME,
};
-use
crate::table::global_index_scanner::unindexed_ranges_for_global_index_entries;
+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,
+};
use crate::table::snapshot_manager::SnapshotManager;
use crate::table::{find_field_id_by_name, merge_row_ranges, RowRange, Table};
use crate::tantivy::full_text_search::{FullTextSearch, SearchResult};
@@ -165,7 +168,8 @@ async fn evaluate_full_text_search(
search: &FullTextSearch,
) -> crate::Result<Vec<RowRange>> {
let table_path = evaluation.table_path.trim_end_matches('/');
- let search_mode =
CoreOptions::new(evaluation.table_options).global_index_search_mode()?;
+ let core_options = CoreOptions::new(evaluation.table_options);
+ let search_mode = core_options.global_index_search_mode()?;
let field_id = match find_field_id_by_name(evaluation.schema_fields,
&search.field_name) {
Some(id) => id,
@@ -189,6 +193,19 @@ async fn evaluate_full_text_search(
return Ok(Vec::new());
}
+ let deleted_row_index = if core_options.data_evolution_enabled() {
+ match evaluation.table {
+ Some(table) => {
+ let ranges =
+ deleted_row_ranges_for_data_evolution_dvs(table,
index_entries).await?;
+ (!ranges.is_empty()).then(|| RowRangeIndex::create(ranges))
+ }
+ None => None,
+ }
+ } else {
+ None
+ };
+
let mut merged = SearchResult::empty();
if !fulltext_entries.is_empty() {
let futures: Vec<_> = fulltext_entries
@@ -198,8 +215,14 @@ async fn evaluate_full_text_search(
let path = format!("{table_path}/{INDEX_DIR}/{}",
entry.index_file.file_name);
let file_name = entry.index_file.file_name.clone();
let query_text = search.query_text.clone();
- let limit = search.limit;
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(),
+ );
let input = evaluation.file_io.new_input(&path);
async move {
let input = input?;
@@ -253,7 +276,10 @@ async fn evaluate_full_text_search(
}
}
- Ok(merged.top_k(search.limit).to_row_ranges())
+ Ok(merged
+ .without_deleted_row_ranges(deleted_row_index.as_ref())?
+ .top_k(search.limit)
+ .to_row_ranges())
}
fn is_tantivy_fulltext_index_file(index_file: &IndexFileMeta) -> bool {
diff --git a/crates/paimon/src/table/global_index_scanner.rs
b/crates/paimon/src/table/global_index_scanner.rs
index 8ed18a9..f604d7c 100644
--- a/crates/paimon/src/table/global_index_scanner.rs
+++ b/crates/paimon/src/table/global_index_scanner.rs
@@ -22,12 +22,13 @@
use crate::btree::query::{extract_between, IndexQuery};
use crate::btree::{make_key_comparator, serialize_datum, BTreeIndexMeta,
BTreeIndexReader};
+use crate::deletion_vector::DeletionVectorFactory;
use crate::io::FileIO;
use crate::spec::{
DataField, DataType, Datum, FileKind, GlobalIndexSearchMode,
IndexFileMeta, IndexManifestEntry,
Predicate, PredicateOperator,
};
-use crate::table::RowRange;
+use crate::table::{DeletionFile, RowRange, Table};
use crate::Result;
use roaring::RoaringTreemap;
use std::cmp::Ordering;
@@ -42,6 +43,7 @@ type EvaluateFuture<'a> =
type PredicateTuple<'a> = (PredicateOperator, &'a [Datum], &'a DataType);
const BTREE_INDEX_TYPE: &str = "btree";
+const DELETION_VECTORS_INDEX_TYPE: &str = "DELETION_VECTORS";
const INDEX_DIR: &str = "index";
/// Evaluates global index predicates and returns matching row ranges.
@@ -663,6 +665,103 @@ pub(crate) fn unindexed_ranges_for_global_index_entries(
)
}
+/// Resolve live deletion-vector index entries into global row-id ranges.
+///
+/// Data-evolution DV entries are keyed by the normal anchor data file. The DV
+/// bitmap positions are local to that anchor file's `first_row_id`, so this
+/// helper joins index metadata with live data-file metadata before converting
+/// deleted positions to global row IDs.
+pub(crate) async fn deleted_row_ranges_for_data_evolution_dvs(
+ table: &Table,
+ index_entries: &[IndexManifestEntry],
+) -> Result<Vec<RowRange>> {
+ if !index_entries.iter().any(|entry| {
+ entry.kind == FileKind::Add && entry.index_file.index_type ==
DELETION_VECTORS_INDEX_TYPE
+ }) {
+ return Ok(Vec::new());
+ }
+
+ let plan = table
+ .new_read_builder()
+ .new_scan()
+ .with_scan_all_files()
+ .plan()
+ .await?;
+
+ let mut first_row_ids: HashMap<(Vec<u8>, i32, String), i64> =
HashMap::new();
+ for split in plan.splits() {
+ let partition = split.partition().to_serialized_bytes();
+ let bucket = split.bucket();
+ for file in split.data_files() {
+ if let Some(first_row_id) = file.first_row_id {
+ first_row_ids.insert(
+ (partition.clone(), bucket, file.file_name.clone()),
+ first_row_id,
+ );
+ }
+ }
+ }
+
+ let mut ranges = Vec::new();
+ let table_path = table.location().trim_end_matches('/');
+ for entry in index_entries {
+ if entry.kind != FileKind::Add || entry.index_file.index_type !=
DELETION_VECTORS_INDEX_TYPE
+ {
+ continue;
+ }
+ let Some(dv_ranges) =
entry.index_file.deletion_vectors_ranges.as_ref() else {
+ continue;
+ };
+ let index_path = format!("{table_path}/{INDEX_DIR}/{}",
entry.index_file.file_name);
+ for (data_file_name, meta) in dv_ranges {
+ let key = (
+ entry.partition.clone(),
+ entry.bucket,
+ data_file_name.clone(),
+ );
+ let first_row_id = first_row_ids.get(&key).copied().ok_or_else(|| {
+ crate::Error::DataInvalid {
+ message: format!(
+ "Deletion vector references data file '{}' but no live
row-tracked file was found",
+ data_file_name
+ ),
+ source: None,
+ }
+ })?;
+ let deletion_file = DeletionFile::new(
+ index_path.clone(),
+ meta.offset as i64,
+ meta.length as i64,
+ meta.cardinality,
+ );
+ let deletion_vector =
+ DeletionVectorFactory::read(table.file_io(),
&deletion_file).await?;
+ for deleted in deletion_vector.iter() {
+ let deleted = i64::try_from(deleted).map_err(|_|
crate::Error::DataInvalid {
+ message: format!(
+ "Deleted position {deleted} for data file '{}' exceeds
i64::MAX",
+ data_file_name
+ ),
+ source: None,
+ })?;
+ let row_id =
+ first_row_id
+ .checked_add(deleted)
+ .ok_or_else(|| crate::Error::DataInvalid {
+ message: format!(
+ "Deleted row id overflows i64 for data file
'{}'",
+ data_file_name
+ ),
+ source: None,
+ })?;
+ ranges.push(RowRange::new(row_id, row_id));
+ }
+ }
+ }
+
+ Ok(super::merge_row_ranges(ranges))
+}
+
/// Index for row ranges. Stores sorted, non-overlapping ranges and supports
/// efficient intersection queries via binary search.
///
@@ -695,12 +794,24 @@ impl RowRangeIndex {
}
/// Returns true if the index has any range that intersects `[start, end]`.
- #[cfg(test)]
pub fn intersects(&self, start: i64, end: i64) -> bool {
let candidate = lower_bound(&self.ends, start);
candidate < self.starts.len() && self.starts[candidate] <= end
}
+ /// Counts rows in this index that intersect `[start, end]`.
+ pub fn intersection_row_count(&self, start: i64, end: i64) -> usize {
+ if start > end {
+ return 0;
+ }
+ self.intersected_ranges(start, end)
+ .into_iter()
+ .fold(0usize, |total, range| {
+ let len =
range.to().saturating_sub(range.from()).saturating_add(1);
+
total.saturating_add(usize::try_from(len).unwrap_or(usize::MAX))
+ })
+ }
+
/// Returns the sub-ranges of this index that intersect `[start, end]`,
/// clipped to the intersection boundaries.
pub fn intersected_ranges(&self, start: i64, end: i64) -> Vec<RowRange> {
@@ -739,6 +850,27 @@ impl RowRangeIndex {
}
}
+pub(crate) fn search_limit_with_deleted_rows(
+ limit: usize,
+ row_range_start: i64,
+ row_range_end: i64,
+ deleted_rows: Option<&RowRangeIndex>,
+) -> usize {
+ let Some(range_len) = row_range_end
+ .checked_sub(row_range_start)
+ .and_then(|len| len.checked_add(1))
+ .and_then(|len| usize::try_from(len).ok())
+ else {
+ return limit;
+ };
+
+ let deleted_count = deleted_rows
+ .map(|index| index.intersection_row_count(row_range_start,
row_range_end))
+ .unwrap_or(0)
+ .min(range_len);
+ limit.saturating_add(deleted_count).min(range_len)
+}
+
/// Binary search: find the first index where `sorted[index] >= target`.
fn lower_bound(sorted: &[i64], target: i64) -> usize {
let mut left = 0;
@@ -904,6 +1036,28 @@ mod tests {
);
}
+ #[test]
+ fn test_row_range_index_intersection_row_count() {
+ let idx = RowRangeIndex::create(vec![
+ RowRange::new(10, 20),
+ RowRange::new(30, 40),
+ RowRange::new(50, 60),
+ ]);
+
+ assert_eq!(idx.intersection_row_count(15, 55), 23);
+ assert_eq!(idx.intersection_row_count(21, 29), 0);
+ assert_eq!(idx.intersection_row_count(55, 15), 0);
+ }
+
+ #[test]
+ fn test_search_limit_with_deleted_rows_expands_and_caps() {
+ let idx = RowRangeIndex::create(vec![RowRange::new(2, 4),
RowRange::new(8, 10)]);
+
+ assert_eq!(search_limit_with_deleted_rows(5, 0, 19, Some(&idx)), 11);
+ assert_eq!(search_limit_with_deleted_rows(18, 0, 19, Some(&idx)), 20);
+ assert_eq!(search_limit_with_deleted_rows(5, 0, 19, None), 5);
+ }
+
#[test]
fn test_row_range_index_empty() {
let idx = RowRangeIndex::create(Vec::new());
diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs
index 1e1a18d..19e9f00 100644
--- a/crates/paimon/src/table/mod.rs
+++ b/crates/paimon/src/table/mod.rs
@@ -70,7 +70,7 @@ use arrow_array::RecordBatch;
pub use branch_manager::BranchManager;
pub use commit_message::CommitMessage;
pub use cow_writer::{CopyOnWriteMergeWriter, FileInfo};
-pub use data_evolution_writer::DataEvolutionWriter;
+pub use data_evolution_writer::{DataEvolutionDeleteWriter,
DataEvolutionWriter};
#[cfg(feature = "fulltext")]
pub use full_text_search_builder::FullTextSearchBuilder;
use futures::stream::BoxStream;
diff --git a/crates/paimon/src/table/source.rs
b/crates/paimon/src/table/source.rs
index 229afa0..7c85d6d 100644
--- a/crates/paimon/src/table/source.rs
+++ b/crates/paimon/src/table/source.rs
@@ -22,6 +22,26 @@
use crate::spec::{BinaryRow, DataFileMeta};
use crate::table::stats_filter::group_by_overlapping_row_id;
use serde::{Deserialize, Serialize};
+
+fn is_vector_store_file_name(file_name: &str) -> bool {
+ file_name.to_ascii_lowercase().contains(".vector.")
+}
+
+pub(crate) fn is_data_evolution_normal_file(file: &DataFileMeta) -> bool {
+ !crate::table::blob_file_writer::is_blob_file_name(&file.file_name)
+ && !is_vector_store_file_name(&file.file_name)
+}
+
+pub(crate) fn data_evolution_anchor_file(files: &[DataFileMeta]) ->
crate::Result<&DataFileMeta> {
+ files
+ .iter()
+ .filter(|file| is_data_evolution_normal_file(file))
+ .min_by_key(|file| (file.max_sequence_number, file.file_name.as_str()))
+ .ok_or_else(|| crate::Error::DataInvalid {
+ message: "Data-evolution deletion vectors require a normal anchor
file in each row range group.".to_string(),
+ source: None,
+ })
+}
// ======================= RowRange ===============================
/// An inclusive row ID range `[from, to]` for filtering reads in data
evolution mode.
@@ -560,10 +580,16 @@ impl DataSplit {
// Merge overlapping row ID ranges and compute max row_count per group
let groups = group_by_overlapping_row_id(self.data_files.to_vec());
- let sum: i64 = groups
+ let mut sum: i64 = groups
.iter()
.map(|group| group.iter().map(|f| f.row_count).max().unwrap_or(0))
.sum();
+ if let Some(deletion_files) = &self.data_deletion_files {
+ for deletion_file in deletion_files.iter().flatten() {
+ sum -= deletion_file.cardinality()?;
+ }
+ }
+
Some(sum)
}
diff --git a/crates/paimon/src/table/table_commit.rs
b/crates/paimon/src/table/table_commit.rs
index 055dcd5..872bea1 100644
--- a/crates/paimon/src/table/table_commit.rs
+++ b/crates/paimon/src/table/table_commit.rs
@@ -43,6 +43,7 @@ use std::time::{SystemTime, UNIX_EPOCH};
const BATCH_COMMIT_IDENTIFIER: i64 = i64::MAX;
/// Java RollingFileWriter.CHECK_ROLLING_RECORD_CNT.
const CHECK_ROLLING_RECORD_COUNT: usize = 1000;
+const DELETION_VECTORS_INDEX_TYPE: &str = "DELETION_VECTORS";
type PartitionBucketKey = (Vec<u8>, i32);
type RowIdRange = (i64, i64);
@@ -1098,6 +1099,12 @@ impl TableCommit {
};
let detect_conflicts = has_delete ||
check_from_snapshot.is_some();
let base_data_files = if detect_conflicts {
+ self.check_deletion_vector_index_only_conflict(
+ latest_snapshot.as_ref(),
+ entries,
+ new_index_entries,
+ *check_from_snapshot,
+ )?;
self.detect_commit_conflicts(
latest_snapshot,
retry_state,
@@ -1709,6 +1716,43 @@ impl TableCommit {
.await
}
+ fn check_deletion_vector_index_only_conflict(
+ &self,
+ latest_snapshot: Option<&Snapshot>,
+ data_entries: &[ManifestEntry],
+ index_entries: &[IndexManifestEntry],
+ check_from_snapshot: Option<i64>,
+ ) -> Result<()> {
+ if !self.data_evolution_enabled || !data_entries.is_empty() {
+ return Ok(());
+ }
+ let Some(check_from_snapshot) = check_from_snapshot else {
+ return Ok(());
+ };
+ let has_deletion_vector_index_change = index_entries
+ .iter()
+ .any(|entry| entry.index_file.index_type ==
DELETION_VECTORS_INDEX_TYPE);
+ if !has_deletion_vector_index_change {
+ return Ok(());
+ }
+ let Some(latest_snapshot) = latest_snapshot else {
+ return Ok(());
+ };
+ if latest_snapshot.id() <= check_from_snapshot {
+ return Ok(());
+ }
+
+ Err(crate::Error::DataInvalid {
+ message: format!(
+ "Row ID conflict: deletion-vector DELETE was prepared from
snapshot \
+ {check_from_snapshot}, but latest snapshot is {}. Retry with
the latest \
+ deletion vectors.",
+ latest_snapshot.id()
+ ),
+ source: None,
+ })
+ }
+
fn check_delete_entries_against_base(
&self,
base_entries: &[ManifestEntry],
@@ -2705,7 +2749,8 @@ mod tests {
use crate::io::FileIOBuilder;
use crate::spec::stats::BinaryTableStats;
use crate::spec::{
- BinaryRowBuilder, DataFileMeta, GlobalIndexMeta, IndexFileMeta,
ManifestList, TableSchema,
+ BinaryRowBuilder, DataFileMeta, DeletionVectorMeta, GlobalIndexMeta,
IndexFileMeta,
+ ManifestList, TableSchema,
};
use chrono::{DateTime, Utc};
@@ -2834,6 +2879,24 @@ mod tests {
file
}
+ fn test_deletion_vector_index_file(name: &str, data_file_name: &str) ->
IndexFileMeta {
+ IndexFileMeta {
+ index_type: DELETION_VECTORS_INDEX_TYPE.to_string(),
+ file_name: name.to_string(),
+ file_size: 128,
+ row_count: 1,
+ deletion_vectors_ranges: Some(indexmap::IndexMap::from([(
+ data_file_name.to_string(),
+ DeletionVectorMeta {
+ offset: 1,
+ length: 16,
+ cardinality: Some(1),
+ },
+ )])),
+ global_index_meta: None,
+ }
+ }
+
fn setup_commit(file_io: &FileIO, table_path: &str) -> TableCommit {
let table = test_table(file_io, table_path);
TableCommit::new(table, "test-user".to_string())
@@ -3143,6 +3206,52 @@ mod tests {
assert!(snapshot.index_manifest().is_none());
}
+ #[tokio::test]
+ async fn test_deletion_vector_index_only_commit_rejects_stale_snapshot() {
+ let file_io = test_file_io();
+ let table_path = "memory:/test_dv_index_only_stale_snapshot";
+ setup_dirs(&file_io, table_path).await;
+
+ let commit = setup_data_evolution_commit(&file_io, table_path);
+ let partition = EMPTY_SERIALIZED_ROW.clone();
+ let mut data_file = test_data_file("data-0.parquet", 10);
+ data_file.file_source = Some(0);
+ commit
+ .commit(vec![CommitMessage::new(
+ partition.clone(),
+ 0,
+ vec![data_file],
+ )])
+ .await
+ .unwrap();
+
+ let mut first_delete = CommitMessage::new(partition.clone(), 0,
vec![]);
+ first_delete.check_from_snapshot = Some(1);
+ first_delete.new_index_files = vec![test_deletion_vector_index_file(
+ "dv-0.index",
+ "data-0.parquet",
+ )];
+ commit.commit(vec![first_delete]).await.unwrap();
+
+ let mut stale_delete = CommitMessage::new(partition, 0, vec![]);
+ stale_delete.check_from_snapshot = Some(1);
+ stale_delete.new_index_files = vec![test_deletion_vector_index_file(
+ "dv-1.index",
+ "data-0.parquet",
+ )];
+ let result = commit.commit(vec![stale_delete]).await;
+
+ assert!(result.is_err());
+ let err_msg = result.unwrap_err().to_string();
+ assert!(
+ err_msg.contains("Row ID conflict"),
+ "expected row-id conflict for stale DV commit, got: {err_msg}"
+ );
+
+ let snapshot = latest_snapshot(&file_io, table_path).await.unwrap();
+ assert_eq!(snapshot.id(), 2);
+ }
+
#[tokio::test]
async fn test_global_index_overlap_rejected_on_commit() {
let file_io = test_file_io();
diff --git a/crates/paimon/src/table/vector_search_builder.rs
b/crates/paimon/src/table/vector_search_builder.rs
index f7df42e..2ca9993 100644
--- a/crates/paimon/src/table/vector_search_builder.rs
+++ b/crates/paimon/src/table/vector_search_builder.rs
@@ -22,7 +22,10 @@ use crate::spec::{
CoreOptions, DataField, FileKind, GlobalIndexSearchMode, IndexFileMeta,
IndexManifest,
IndexManifestEntry, ROW_ID_FIELD_NAME,
};
-use
crate::table::global_index_scanner::unindexed_ranges_for_global_index_entries;
+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,
+};
use crate::table::snapshot_manager::SnapshotManager;
use crate::table::{find_field_id_by_name, merge_row_ranges, RowRange, Table};
use crate::vector_search::{GlobalIndexIOMeta, SearchResult, VectorSearch};
@@ -264,7 +267,8 @@ async fn evaluate_batch_vector_search(
}
let table_path = evaluation.table_path.trim_end_matches('/');
- let search_mode =
CoreOptions::new(evaluation.table_options).global_index_search_mode()?;
+ let core_options = CoreOptions::new(evaluation.table_options);
+ let search_mode = core_options.global_index_search_mode()?;
let field_name = &vector_searches[0].field_name;
if vector_searches
.iter()
@@ -298,6 +302,19 @@ async fn evaluate_batch_vector_search(
return Ok(vec![SearchResult::empty(); vector_searches.len()]);
}
+ let deleted_row_index = if core_options.data_evolution_enabled() {
+ match evaluation.table {
+ Some(table) => {
+ let ranges =
+ deleted_row_ranges_for_data_evolution_dvs(table,
index_entries).await?;
+ (!ranges.is_empty()).then(|| RowRangeIndex::create(ranges))
+ }
+ None => None,
+ }
+ } else {
+ None
+ };
+
let mut merged = vec![SearchResult::empty(); vector_searches.len()];
if !vector_entries.is_empty() {
let futures: Vec<_> = vector_entries
@@ -311,7 +328,22 @@ async fn evaluate_batch_vector_search(
let file_size = entry.index_file.file_size as u64;
let index_meta_bytes =
global_meta.index_meta.clone().unwrap_or_default();
let row_range_start = global_meta.row_range_start;
- let vector_searches = vector_searches.to_vec();
+ let row_range_end = global_meta.row_range_end;
+ let max_limit = vector_searches
+ .iter()
+ .map(|vector_search| vector_search.limit)
+ .max()
+ .unwrap_or(0);
+ let index_limit = search_limit_with_deleted_rows(
+ max_limit,
+ row_range_start,
+ row_range_end,
+ deleted_row_index.as_ref(),
+ );
+ let mut vector_searches = vector_searches.to_vec();
+ for vector_search in &mut vector_searches {
+ vector_search.limit = index_limit;
+ }
let options = evaluation.table_options.clone();
let input = evaluation.file_io.new_input(&path);
async move {
@@ -417,11 +449,15 @@ async fn evaluate_batch_vector_search(
}
}
- Ok(merged
+ merged
.into_iter()
.zip(vector_searches)
- .map(|(result, vector_search)| result.top_k(vector_search.limit))
- .collect())
+ .map(|(result, vector_search)| {
+ Ok(result
+ .without_deleted_row_ranges(deleted_row_index.as_ref())?
+ .top_k(vector_search.limit))
+ })
+ .collect()
}
fn is_vector_global_index_file(index_file: &IndexFileMeta) -> bool {
diff --git a/crates/paimon/src/table/write_builder.rs
b/crates/paimon/src/table/write_builder.rs
index 34ae27d..703fd2c 100644
--- a/crates/paimon/src/table/write_builder.rs
+++ b/crates/paimon/src/table/write_builder.rs
@@ -19,7 +19,7 @@
//!
//! Reference: [pypaimon
WriteBuilder](https://github.com/apache/paimon/blob/master/paimon-python/pypaimon/write/write_builder.py)
-use crate::table::{Table, TableCommit, TableUpdate, TableWrite};
+use crate::table::{DataEvolutionDeleteWriter, Table, TableCommit, TableUpdate,
TableWrite};
use uuid::Uuid;
/// Builder for creating table writers and committers.
@@ -109,6 +109,11 @@ impl<'a> WriteBuilder<'a> {
pub fn new_update(&self, update_columns: Vec<String>) ->
crate::Result<TableUpdate> {
TableUpdate::new(self.table, update_columns)
}
+
+ /// Create a new writer for data-evolution row-id deletes.
+ pub fn new_delete(&self) -> crate::Result<DataEvolutionDeleteWriter> {
+ DataEvolutionDeleteWriter::new(self.table)
+ }
}
fn validate_commit_user(commit_user: &str) -> crate::Result<()> {
diff --git a/crates/paimon/src/tantivy/full_text_search.rs
b/crates/paimon/src/tantivy/full_text_search.rs
index 77bcd03..2f57ef8 100644
--- a/crates/paimon/src/tantivy/full_text_search.rs
+++ b/crates/paimon/src/tantivy/full_text_search.rs
@@ -130,6 +130,31 @@ impl SearchResult {
Self { row_ids, scores }
}
+ pub(crate) fn without_deleted_row_ranges(
+ &self,
+ deleted_rows:
Option<&crate::table::global_index_scanner::RowRangeIndex>,
+ ) -> crate::Result<Self> {
+ let Some(deleted_rows) = deleted_rows else {
+ return Ok(self.clone());
+ };
+
+ let mut row_ids = Vec::with_capacity(self.row_ids.len());
+ let mut scores = Vec::with_capacity(self.scores.len());
+ for (&row_id, &score) in self.row_ids.iter().zip(&self.scores) {
+ let row_id_i64 = i64::try_from(row_id).map_err(|_|
crate::Error::DataInvalid {
+ message: format!(
+ "Full-text search row id {row_id} exceeds i64::MAX and
cannot be checked against deletion vectors"
+ ),
+ source: None,
+ })?;
+ if !deleted_rows.intersects(row_id_i64, row_id_i64) {
+ row_ids.push(row_id);
+ scores.push(score);
+ }
+ }
+ Ok(Self { row_ids, scores })
+ }
+
/// Convert to sorted, merged row ranges.
pub fn to_row_ranges(&self) -> Vec<crate::table::RowRange> {
if self.row_ids.is_empty() {
@@ -173,4 +198,19 @@ mod tests {
let result = FullTextSearch::new("".into(), 10, "text".into());
assert!(result.is_err());
}
+
+ #[test]
+ fn test_search_result_filters_deleted_row_ranges() {
+ let result = SearchResult::new(vec![1, 2, 3, 4], vec![0.1, 0.9, 0.8,
0.2]);
+ let deleted =
crate::table::global_index_scanner::RowRangeIndex::create(vec![
+ crate::table::RowRange::new(2, 3),
+ ]);
+
+ let filtered = result
+ .without_deleted_row_ranges(Some(&deleted))
+ .unwrap()
+ .top_k(10);
+ assert_eq!(filtered.row_ids, vec![1, 4]);
+ assert_eq!(filtered.scores, vec![0.1, 0.2]);
+ }
}
diff --git a/crates/paimon/src/vector_search.rs
b/crates/paimon/src/vector_search.rs
index 491b65f..28b6500 100644
--- a/crates/paimon/src/vector_search.rs
+++ b/crates/paimon/src/vector_search.rs
@@ -167,6 +167,31 @@ impl SearchResult {
Self { row_ids, scores }
}
+ pub(crate) fn without_deleted_row_ranges(
+ &self,
+ deleted_rows:
Option<&crate::table::global_index_scanner::RowRangeIndex>,
+ ) -> crate::Result<Self> {
+ let Some(deleted_rows) = deleted_rows else {
+ return Ok(self.clone());
+ };
+
+ let mut row_ids = Vec::with_capacity(self.row_ids.len());
+ let mut scores = Vec::with_capacity(self.scores.len());
+ for (&row_id, &score) in self.row_ids.iter().zip(&self.scores) {
+ let row_id_i64 = i64::try_from(row_id).map_err(|_|
crate::Error::DataInvalid {
+ message: format!(
+ "Vector search row id {row_id} exceeds i64::MAX and cannot
be checked against deletion vectors"
+ ),
+ source: None,
+ })?;
+ if !deleted_rows.intersects(row_id_i64, row_id_i64) {
+ row_ids.push(row_id);
+ scores.push(score);
+ }
+ }
+ Ok(Self { row_ids, scores })
+ }
+
pub fn to_row_ranges(&self) -> crate::Result<Vec<crate::table::RowRange>> {
if self.row_ids.is_empty() {
return Ok(Vec::new());
@@ -244,6 +269,21 @@ mod tests {
assert!(top.row_ids.contains(&4));
}
+ #[test]
+ fn test_search_result_filters_deleted_row_ranges() {
+ let result = SearchResult::new(vec![1, 2, 3, 4], vec![0.1, 0.9, 0.8,
0.2]);
+ let deleted =
crate::table::global_index_scanner::RowRangeIndex::create(vec![
+ crate::table::RowRange::new(2, 3),
+ ]);
+
+ let filtered = result
+ .without_deleted_row_ranges(Some(&deleted))
+ .unwrap()
+ .top_k(10);
+ assert_eq!(filtered.row_ids, vec![1, 4]);
+ assert_eq!(filtered.scores, vec![0.1, 0.2]);
+ }
+
#[test]
fn test_search_result_offset() {
let result = SearchResult::new(vec![0, 1], vec![0.5, 0.6]);
