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 f282f586 [Core] Add predicate deletes to TableUpdate (#1105)
f282f586 is described below
commit f282f586d0be99938036dd02df028a8eeed7dbd9
Author: Jingsong Lee <[email protected]>
AuthorDate: Mon Oct 12 08:21:09 2026 +0800
[Core] Add predicate deletes to TableUpdate (#1105)
---
.../python/python/pypaimon_rust/datafusion.pyi | 8 +
bindings/python/src/write.rs | 38 ++
bindings/python/tests/test_predicate_delete.py | 68 +++
crates/paimon/src/table/mod.rs | 2 +
crates/paimon/src/table/snapshot_manifest.rs | 49 ++
crates/paimon/src/table/table_commit.rs | 24 +-
crates/paimon/src/table/table_update.rs | 16 +
crates/paimon/src/table/table_update_delete.rs | 164 ++++++
crates/paimon/tests/table_predicate_delete_test.rs | 573 +++++++++++++++++++++
9 files changed, 924 insertions(+), 18 deletions(-)
diff --git a/bindings/python/python/pypaimon_rust/datafusion.pyi
b/bindings/python/python/pypaimon_rust/datafusion.pyi
index e0414953..b14bbe5b 100644
--- a/bindings/python/python/pypaimon_rust/datafusion.pyi
+++ b/bindings/python/python/pypaimon_rust/datafusion.pyi
@@ -345,6 +345,11 @@ class BatchTableUpdate:
...
def with_update_type(self, update_cols: List[str]) -> "BatchTableUpdate":
...
def delete_by_row_id(self, row_ids: Sequence[int]) -> List[CommitMessage]:
...
+ def delete_by_predicate(
+ self, predicate: Optional[Dict[str, Any]]
+ ) -> List[CommitMessage]:
+ """Prepare partition metadata deletes or matching row deletion vectors
in core."""
+ ...
def upsert_by_arrow_with_key(
self, table: Union[pyarrow.Table, Sequence[pyarrow.RecordBatch]],
upsert_keys: List[str]
) -> List[CommitMessage]:
@@ -399,6 +404,9 @@ class StreamTableUpdate:
def delete_by_row_id(
self, row_ids: Sequence[int], commit_identifier: int
) -> List[CommitMessage]: ...
+ def delete_by_predicate(
+ self, predicate: Optional[Dict[str, Any]], commit_identifier: int
+ ) -> List[CommitMessage]: ...
def upsert_by_arrow_with_key(
self, table: Union[pyarrow.Table, Sequence[pyarrow.RecordBatch]],
upsert_keys: List[str], commit_identifier: int
diff --git a/bindings/python/src/write.rs b/bindings/python/src/write.rs
index fffe64b9..be06b18f 100644
--- a/bindings/python/src/write.rs
+++ b/bindings/python/src/write.rs
@@ -788,6 +788,24 @@ impl UpdateContext {
))
}
+ fn delete_by_predicate(
+ &self,
+ py: Python<'_>,
+ predicate: Option<&Bound<'_, PyDict>>,
+ ) -> PyResult<Vec<PyCommitMessage>> {
+ let predicate = predicate
+ .map(|predicate| dict_to_table_predicate(predicate,
self.table.schema(), true))
+ .transpose()?;
+ let messages = py
+ .detach(||
runtime().block_on(self.inner.delete_by_predicate(predicate)))
+ .map_err(to_py_err)?;
+ Ok(wrap_messages(
+ messages,
+ &self.table_location,
+ &self.commit_user,
+ ))
+ }
+
fn delete_by_row_id(
&self,
py: Python<'_>,
@@ -1058,10 +1076,30 @@ impl PyStreamTableUpdate {
let _ = commit_identifier;
self.context.delete_by_row_id(py, row_ids)
}
+
+ #[pyo3(signature = (predicate, commit_identifier))]
+ fn delete_by_predicate(
+ &self,
+ py: Python<'_>,
+ predicate: Option<&Bound<'_, PyDict>>,
+ commit_identifier: i64,
+ ) -> PyResult<Vec<PyCommitMessage>> {
+ let _ = commit_identifier;
+ self.context.delete_by_predicate(py, predicate)
+ }
}
#[pymethods]
impl PyBatchTableUpdate {
+ #[pyo3(signature = (predicate))]
+ fn delete_by_predicate(
+ &self,
+ py: Python<'_>,
+ predicate: Option<&Bound<'_, PyDict>>,
+ ) -> PyResult<Vec<PyCommitMessage>> {
+ self.context.delete_by_predicate(py, predicate)
+ }
+
fn _with_blob_uri_reader_factory(
&mut self,
py: Python<'_>,
diff --git a/bindings/python/tests/test_predicate_delete.py
b/bindings/python/tests/test_predicate_delete.py
new file mode 100644
index 00000000..1e27fbcb
--- /dev/null
+++ b/bindings/python/tests/test_predicate_delete.py
@@ -0,0 +1,68 @@
+# 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.
+
+import pyarrow as pa
+import pytest
+
+from pypaimon_rust.datafusion import PaimonCatalog, SQLContext
+
+
[email protected]('stream', [False, True])
[email protected]('partition_delete', [False, True])
+def test_predicate_delete_returns_messages_for_batch_and_stream(tmp_path,
stream, partition_delete):
+ context = SQLContext()
+ context.register_catalog('paimon', {'warehouse': str(tmp_path)})
+ context.sql('CREATE SCHEMA paimon.deletes')
+ context.sql("""CREATE TABLE paimon.deletes.t (id INT, p STRING, v INT)
PARTITIONED BY (p) WITH (
+ 'row-tracking.enabled' = 'true', 'data-evolution.enabled' = 'true',
+ 'deletion-vectors.enabled' = '%s')""" % ('false' if partition_delete
else 'true'))
+ context.sql("INSERT INTO paimon.deletes.t (id, p, v) VALUES (1, 'a', 10),
(2, 'a', 20), (3, 'b', 30)")
+ table = PaimonCatalog({'warehouse': str(tmp_path)}).get_table('deletes.t')
+ builder =
(table.new_stream_write_builder().with_commit_user('predicate-stream') if stream
+ else table.new_batch_write_builder())
+ predicate = {'method': 'equal', 'field': 'p' if partition_delete else 'v',
+ 'literals': ['a' if partition_delete else 20]}
+ update = builder.new_update()
+ messages = (update.delete_by_predicate(predicate, 7) if stream
+ else update.delete_by_predicate(predicate))
+ assert messages and all(message.serialize() for message in messages)
+ # Preparation alone does not publish the delete.
+ assert pa.Table.from_batches(context.sql('SELECT id FROM
paimon.deletes.t')).num_rows == 3
+ commit = builder.new_commit()
+ commit.commit(7, messages) if stream else commit.commit(messages)
+ assert pa.Table.from_batches(context.sql(
+ 'SELECT id FROM paimon.deletes.t ORDER BY id'))['id'].to_pylist() == (
+ [3] if partition_delete else [1, 3])
+
+
+def
test_predicate_delete_requires_explicit_selection_but_none_matches_all(tmp_path):
+ context = SQLContext()
+ context.register_catalog('paimon', {'warehouse': str(tmp_path)})
+ context.sql('CREATE SCHEMA paimon.deletes')
+ context.sql("""CREATE TABLE paimon.deletes.t (id INT) WITH (
+ 'row-tracking.enabled' = 'true', 'data-evolution.enabled' = 'true',
+ 'deletion-vectors.enabled' = 'true')""")
+ context.sql('INSERT INTO paimon.deletes.t (id) VALUES (1), (2)')
+ table = PaimonCatalog({'warehouse': str(tmp_path)}).get_table('deletes.t')
+ builder = table.new_batch_write_builder()
+ update = builder.new_update()
+ with pytest.raises(TypeError, match='predicate'):
+ update.delete_by_predicate()
+ messages = update.delete_by_predicate(None)
+ assert messages
+ builder.new_commit().commit(messages)
+ assert list(context.sql('SELECT id FROM paimon.deletes.t')) == []
diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs
index cf1f164a..fe7f8ced 100644
--- a/crates/paimon/src/table/mod.rs
+++ b/crates/paimon/src/table/mod.rs
@@ -140,6 +140,7 @@ pub(crate) mod schema_manager;
mod shard_table_updater;
pub(crate) mod snapshot_commit;
mod snapshot_manager;
+mod snapshot_manifest;
mod sort_merge;
mod sorted_global_index_build_builder;
mod sorted_global_index_options;
@@ -155,6 +156,7 @@ mod table_read;
mod table_scan;
mod table_update;
mod table_update_by_row_id;
+mod table_update_delete;
mod table_update_predicate;
mod table_upsert;
pub(crate) mod table_write;
diff --git a/crates/paimon/src/table/snapshot_manifest.rs
b/crates/paimon/src/table/snapshot_manifest.rs
new file mode 100644
index 00000000..79391880
--- /dev/null
+++ b/crates/paimon/src/table/snapshot_manifest.rs
@@ -0,0 +1,49 @@
+// 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.
+
+use super::partition_filter::PartitionFilter;
+use super::SnapshotManager;
+use crate::spec::{merge_active_entries, Manifest, ManifestEntry, ManifestList,
Snapshot};
+
+/// Read active metadata without read-side bucket, level, column, or index
pruning.
+/// Partition mutations must also see pending files and dedicated column files.
+pub(super) async fn read_snapshot_entries(
+ manager: &SnapshotManager,
+ snapshot: &Snapshot,
+ partition_filter: Option<&PartitionFilter>,
+) -> crate::Result<Vec<ManifestEntry>> {
+ let file_io = manager.file_io();
+ let mut entries = Vec::new();
+ for list in [
+ snapshot.base_manifest_list(),
+ snapshot.delta_manifest_list(),
+ ] {
+ for manifest in ManifestList::read(file_io,
&manager.manifest_path(list)).await? {
+ for entry in
+ Manifest::read(file_io,
&manager.manifest_path(manifest.file_name())).await?
+ {
+ if let Some(filter) = partition_filter {
+ if !filter.matches_entry(entry.partition())? {
+ continue;
+ }
+ }
+ entries.push(entry);
+ }
+ }
+ }
+ Ok(merge_active_entries(entries))
+}
diff --git a/crates/paimon/src/table/table_commit.rs
b/crates/paimon/src/table/table_commit.rs
index d4d2dba8..317ac221 100644
--- a/crates/paimon/src/table/table_commit.rs
+++ b/crates/paimon/src/table/table_commit.rs
@@ -2270,24 +2270,12 @@ impl TableCommit {
let Some(snap) = snapshot else {
return Ok(vec![]);
};
- let file_io = self.snapshot_manager.file_io();
- let manifest_dir = self.snapshot_manager.manifest_dir();
- let mut entries = Vec::new();
- for manifest_list in [snap.base_manifest_list(),
snap.delta_manifest_list()] {
- let manifest_list_path = format!("{manifest_dir}/{manifest_list}");
- for manifest_file in ManifestList::read(file_io,
&manifest_list_path).await? {
- let manifest_path = format!("{manifest_dir}/{}",
manifest_file.file_name());
- for entry in Manifest::read(file_io, &manifest_path).await? {
- if let Some(filter) = partition_filter {
- if !filter.matches_entry(entry.partition())? {
- continue;
- }
- }
- entries.push(entry);
- }
- }
- }
- Ok(merge_active_entries(entries))
+ super::snapshot_manifest::read_snapshot_entries(
+ &self.snapshot_manager,
+ snap,
+ partition_filter,
+ )
+ .await
}
async fn scan_changed_partition_entries(
diff --git a/crates/paimon/src/table/table_update.rs
b/crates/paimon/src/table/table_update.rs
index af9b283a..546dc8a2 100644
--- a/crates/paimon/src/table/table_update.rs
+++ b/crates/paimon/src/table/table_update.rs
@@ -288,6 +288,22 @@ impl TableUpdate {
.await
}
+ /// Delete rows matching a predicate against one captured snapshot.
+ ///
+ /// Partition-only predicates remove complete data and index metadata,
+ /// including pending files, for append and primary-key tables as well as
+ /// data-evolution tables, without requiring deletion vectors. All other
+ /// predicates (including `None`, which matches all rows) require a
+ /// data-evolution table with row tracking and deletion vectors.
+ /// Returned messages carry the snapshot baseline for conflict detection;
+ /// the caller publishes them through the matching builder's committer.
+ pub async fn delete_by_predicate(
+ &self,
+ predicate: Option<crate::spec::Predicate>,
+ ) -> crate::Result<Vec<CommitMessage>> {
+ super::table_update_delete::delete(&self.table, predicate).await
+ }
+
/// Delete rows by row ID using deletion vectors.
pub async fn delete_by_row_id(&self, row_ids: Vec<i64>) ->
crate::Result<Vec<CommitMessage>> {
if row_ids.is_empty() {
diff --git a/crates/paimon/src/table/table_update_delete.rs
b/crates/paimon/src/table/table_update_delete.rs
new file mode 100644
index 00000000..84b3af40
--- /dev/null
+++ b/crates/paimon/src/table/table_update_delete.rs
@@ -0,0 +1,164 @@
+// 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.
+
+//! Predicate deletes prepare changes against one captured snapshot.
+
+use std::collections::HashMap;
+
+use futures::TryStreamExt;
+use indexmap::IndexMap;
+
+use super::partition_filter::PartitionFilter;
+use super::{CommitMessage, DataEvolutionDeleteWriter, Table};
+use crate::spec::{
+ field_idx_to_partition_idx, CoreOptions, FileKind, IndexManifest,
Predicate, Snapshot,
+};
+
+pub(super) async fn delete(
+ table: &Table,
+ predicate: Option<Predicate>,
+) -> crate::Result<Vec<CommitMessage>> {
+ super::write_builder::ensure_table_write_allowed(table)?;
+ let options = CoreOptions::new(table.schema().options());
+ options.ensure_read_authorized()?;
+ if let Some(predicate) = &predicate {
+ validate_predicate(predicate,
&table.schema().fields_with_read_metadata())?;
+ }
+
+ let mapping =
+ field_idx_to_partition_idx(table.schema().fields(),
table.schema().partition_keys());
+ let partition_predicate = predicate.as_ref().and_then(|predicate| {
+ (!table.schema().partition_keys().is_empty())
+ .then(|| predicate.remap_field_index(&mapping))
+ .flatten()
+ });
+ // Data predicates require deletion vectors even when the table is empty or
+ // no rows match. Partition-only deletes remove complete file/index
metadata.
+ if partition_predicate.is_none() {
+ let _ = DataEvolutionDeleteWriter::new(table)?;
+ }
+ let Some(snapshot) = super::time_travel::resolve_snapshot(table).await?
else {
+ return Ok(Vec::new());
+ };
+ if let Some(predicate) = partition_predicate {
+ return delete_partitions(table, &snapshot, predicate).await;
+ }
+
+ // FULL includes unindexed rows in PARTIAL/DETAIL search mode. Both the
+ // predicate reader and the DV writer must keep this exact snapshot.
+ let scan_table = table
+ .copy_with_options(HashMap::from([(
+ "scalar-index.search-mode".into(),
+ "FULL".into(),
+ )]))
+ .copy_with_pinned_snapshot(Some(&snapshot));
+ let mut builder = scan_table.new_read_builder();
+ builder.with_projection(&["_ROW_ID"])?;
+ if let Some(predicate) = predicate {
+ builder.with_filter(predicate);
+ }
+ let reader = builder.new_read()?;
+ let plan = builder.new_scan().with_scan_all_files().plan().await?;
+ let mut writer = DataEvolutionDeleteWriter::new(&scan_table)?;
+ let mut batches = reader.to_arrow(plan.splits())?;
+ while let Some(batch) = batches.try_next().await? {
+ writer.add_matched_batch(batch)?;
+ }
+ writer.prepare_commit().await
+}
+
+fn validate_predicate(
+ predicate: &Predicate,
+ fields: &[crate::spec::DataField],
+) -> crate::Result<()> {
+ match predicate {
+ Predicate::Leaf {
+ column,
+ index,
+ data_type,
+ ..
+ } => {
+ if !fields
+ .get(*index)
+ .is_some_and(|field| field.name() == column &&
field.data_type() == data_type)
+ {
+ return Err(crate::Error::DataInvalid {
+ message: format!("Delete predicate field '{column}' does
not match the table schema at index {index}"),
+ source: None,
+ });
+ }
+ }
+ Predicate::And(children) | Predicate::Or(children) => {
+ for child in children {
+ validate_predicate(child, fields)?;
+ }
+ }
+ Predicate::Not(child) => validate_predicate(child, fields)?,
+ Predicate::AlwaysTrue | Predicate::AlwaysFalse => {}
+ }
+ Ok(())
+}
+
+async fn delete_partitions(
+ table: &Table,
+ snapshot: &Snapshot,
+ predicate: Predicate,
+) -> crate::Result<Vec<CommitMessage>> {
+ let partition_fields = table.schema().partition_fields();
+ let filter = PartitionFilter::from_predicate(predicate, &partition_fields);
+ let manager = table.snapshot_manager();
+ let mut messages = IndexMap::new();
+ for entry in
+ super::snapshot_manifest::read_snapshot_entries(&manager, snapshot,
Some(&filter)).await?
+ {
+ partition_message(
+ &mut messages,
+ entry.partition(),
+ entry.bucket(),
+ snapshot.id(),
+ )
+ .deleted_files
+ .push(entry.file().clone());
+ }
+ if let Some(index_manifest) = snapshot.index_manifest() {
+ for entry in
+ IndexManifest::read(table.file_io(),
&manager.manifest_path(index_manifest)).await?
+ {
+ if entry.kind == FileKind::Add &&
filter.matches_entry(&entry.partition)? {
+ partition_message(&mut messages, &entry.partition,
entry.bucket, snapshot.id())
+ .deleted_index_files
+ .push(entry.index_file);
+ }
+ }
+ }
+ Ok(messages.into_values().collect())
+}
+
+fn partition_message<'a>(
+ messages: &'a mut IndexMap<(Vec<u8>, i32), CommitMessage>,
+ partition: &[u8],
+ bucket: i32,
+ snapshot_id: i64,
+) -> &'a mut CommitMessage {
+ messages
+ .entry((partition.to_vec(), bucket))
+ .or_insert_with(|| {
+ let mut message = CommitMessage::new(partition.to_vec(), bucket,
Vec::new());
+ message.check_from_snapshot = Some(snapshot_id);
+ message
+ })
+}
diff --git a/crates/paimon/tests/table_predicate_delete_test.rs
b/crates/paimon/tests/table_predicate_delete_test.rs
new file mode 100644
index 00000000..cc30f344
--- /dev/null
+++ b/crates/paimon/tests/table_predicate_delete_test.rs
@@ -0,0 +1,573 @@
+// 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.
+
+mod common;
+
+use std::collections::HashMap;
+use std::sync::Arc;
+
+use arrow_array::{ArrayRef, Int32Array, Int64Array, RecordBatch, StringArray};
+use common::incremental_helpers::{memory_table, persist_table_schema,
setup_dirs, write_batch};
+use futures::TryStreamExt;
+use paimon::spec::{
+ DataType, Datum, IndexManifest, IntType, Predicate, PredicateBuilder,
Schema, TableSchema,
+ VarCharType,
+};
+use paimon::table::{CommitMessage, Table};
+
+async fn table(options: &[(&str, &str)]) -> Table {
+ table_with_primary_keys(options, false).await
+}
+
+async fn table_with_primary_keys(options: &[(&str, &str)], pk: bool) -> Table {
+ let mut schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("p", DataType::VarChar(VarCharType::string_type()))
+ .column("v", DataType::Int(IntType::new()))
+ .column("q", DataType::Int(IntType::new()))
+ .partition_keys(["q", "p"])
+ .primary_key(if pk { vec!["id", "q", "p"] } else { vec![] })
+ .option("data-evolution.enabled", "true")
+ .option("row-tracking.enabled", "true")
+ .option("deletion-vectors.enabled", "true");
+ for (key, value) in options {
+ schema = schema.option(*key, *value);
+ }
+ let path = "memory:/predicate_delete";
+ let (io, table) = memory_table(path, TableSchema::new(0,
&schema.build().unwrap()));
+ setup_dirs(&io, path).await;
+ persist_table_schema(&io, path, table.schema()).await;
+ table
+}
+
+fn batch(id: Vec<i32>, p: Vec<Option<&str>>, v: Vec<i32>, q: Vec<i32>) ->
RecordBatch {
+ RecordBatch::try_from_iter([
+ ("id", Arc::new(Int32Array::from(id)) as ArrayRef),
+ ("p", Arc::new(StringArray::from(p)) as ArrayRef),
+ ("v", Arc::new(Int32Array::from(v)) as ArrayRef),
+ ("q", Arc::new(Int32Array::from(q)) as ArrayRef),
+ ])
+ .unwrap()
+}
+
+async fn seed(table: &Table) {
+ write_batch(
+ table,
+ &batch(
+ vec![0, 1, 2, 3],
+ vec![
+ Some("a"),
+ Some("a"),
+ Some("b"),
+ if table.schema().primary_keys().is_empty() {
+ None
+ } else {
+ Some("c")
+ },
+ ],
+ vec![10, 20, 30, 40],
+ vec![1, 1, 2, 2],
+ ),
+ )
+ .await;
+}
+
+async fn commit(table: &Table, messages: Vec<CommitMessage>) {
+ table
+ .new_write_builder()
+ .new_commit()
+ .commit(messages)
+ .await
+ .unwrap();
+}
+
+async fn ids(table: &Table) -> Vec<i32> {
+ // Ordinary FULL reads let the assertion include unindexed rows even if
+ // the operation under test was configured for PARTIAL/DETAIL search.
+ let table = table.copy_with_options(HashMap::from([(
+ "scalar-index.search-mode".into(),
+ "FULL".into(),
+ )]));
+ let mut builder = table.new_read_builder();
+ builder.with_projection(&["id"]).unwrap();
+ let plan = builder.new_scan().plan().await.unwrap();
+ let batches: Vec<RecordBatch> = builder
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ let mut ids: Vec<_> = batches
+ .iter()
+ .flat_map(|batch| {
+ batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap()
+ .values()
+ .to_vec()
+ })
+ .collect();
+ ids.sort_unstable();
+ ids
+}
+
+async fn snapshot_id(table: &Table) -> i64 {
+ table
+ .snapshot_manager()
+ .get_latest_snapshot()
+ .await
+ .unwrap()
+ .unwrap()
+ .id()
+}
+
+async fn index_names(table: &Table) -> Vec<String> {
+ let snapshot = table
+ .snapshot_manager()
+ .get_latest_snapshot()
+ .await
+ .unwrap()
+ .unwrap();
+ let Some(name) = snapshot.index_manifest() else {
+ return Vec::new();
+ };
+ let mut names: Vec<_> = IndexManifest::read(
+ table.file_io(),
+ &table.snapshot_manager().manifest_path(name),
+ )
+ .await
+ .unwrap()
+ .into_iter()
+ .map(|entry| entry.index_file.file_name)
+ .collect();
+ names.sort();
+ names
+}
+
+#[tokio::test]
+async fn
data_predicate_uses_latest_partial_column_values_and_existing_deletes() {
+ let table = table(&[("read.batch-size", "1")]).await;
+ seed(&table).await;
+ let fields = table.schema().fields_with_read_metadata();
+ let pb = PredicateBuilder::new(&fields);
+ let mut read = table.new_read_builder();
+ read.with_projection(&["id", "_ROW_ID"]).unwrap();
+ let plan = read.new_scan().plan().await.unwrap();
+ let batches: Vec<RecordBatch> = read
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ let mut row_ids = HashMap::new();
+ for batch in batches {
+ let ids = batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ let rows = batch
+ .column(1)
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .unwrap();
+ for row in 0..batch.num_rows() {
+ row_ids.insert(ids.value(row), rows.value(row));
+ }
+ }
+ let update = table.new_write_builder().new_update().unwrap();
+ let updates = RecordBatch::try_from_iter([
+ (
+ "_ROW_ID",
+ Arc::new(Int64Array::from(vec![row_ids[&0]])) as ArrayRef,
+ ),
+ ("v", Arc::new(Int32Array::from(vec![99])) as ArrayRef),
+ ])
+ .unwrap();
+ commit(
+ &table,
+ update
+ .update_by_arrow_with_row_id(vec![updates])
+ .await
+ .unwrap(),
+ )
+ .await;
+ commit(
+ &table,
+ update.delete_by_row_id(vec![row_ids[&1]]).await.unwrap(),
+ )
+ .await;
+ let baseline = snapshot_id(&table).await;
+ let messages = update
+ .delete_by_predicate(Some(pb.greater_than("v",
Datum::Int(50)).unwrap()))
+ .await
+ .unwrap();
+ assert_eq!(messages.len(), 1);
+ assert_eq!(messages[0].check_from_snapshot, Some(baseline));
+ assert!(messages[0].deleted_files.is_empty());
+ assert_eq!(messages[0].new_index_files.len(), 1);
+ assert_eq!(messages[0].deleted_index_files.len(), 1);
+ assert_eq!(ids(&table).await, vec![0, 2, 3]);
+ commit(&table, messages).await;
+ assert_eq!(ids(&table).await, vec![2, 3]);
+ assert!(update
+ .delete_by_predicate(Some(pb.greater_than("v",
Datum::Int(50)).unwrap()))
+ .await
+ .unwrap()
+ .is_empty());
+}
+
+#[tokio::test]
+async fn partition_predicates_remap_all_leaves_and_delete_without_dv() {
+ for (predicate, expected) in [
+ (0, vec![2]),
+ (1, vec![2, 3]),
+ (2, vec![0, 1, 2]),
+ (3, vec![0, 1]),
+ ] {
+ let table = table(&[("deletion-vectors.enabled", "false")]).await;
+ seed(&table).await;
+ let pb = PredicateBuilder::new(table.schema().fields());
+ let predicate = match predicate {
+ 0 => Predicate::or(vec![
+ pb.equal("p", Datum::String("a".into())).unwrap(),
+ pb.is_null("p").unwrap(),
+ ]),
+ 1 => Predicate::and(vec![
+ pb.equal("q", Datum::Int(1)).unwrap(),
+ pb.equal("p", Datum::String("a".into())).unwrap(),
+ ]),
+ 2 => pb.is_null("p").unwrap(),
+ _ => Predicate::negate(pb.equal("q", Datum::Int(1)).unwrap()),
+ };
+ let baseline = snapshot_id(&table).await;
+ let messages = table
+ .new_write_builder()
+ .new_update()
+ .unwrap()
+ .delete_by_predicate(Some(predicate))
+ .await
+ .unwrap();
+ assert!(messages
+ .iter()
+ .all(|message| message.check_from_snapshot == Some(baseline)
+ && message.new_index_files.is_empty()
+ && !message.deleted_files.is_empty()));
+ assert_eq!(ids(&table).await, vec![0, 1, 2, 3]);
+ commit(&table, messages).await;
+ assert_eq!(ids(&table).await, expected);
+ }
+}
+
+#[tokio::test]
+async fn mixed_and_or_and_not_predicates_delete_rows_not_whole_partitions() {
+ for predicate_kind in 0..3 {
+ let table = table(&[]).await;
+ seed(&table).await;
+ let pb = PredicateBuilder::new(table.schema().fields());
+ let partition = pb.equal("p", Datum::String("a".into())).unwrap();
+ let data = pb.equal("v", Datum::Int(20)).unwrap();
+ let (predicate, expected) = match predicate_kind {
+ 0 => (Predicate::and(vec![partition, data]), vec![0, 2, 3]),
+ 1 => (
+ Predicate::or(vec![partition, pb.equal("v",
Datum::Int(40)).unwrap()]),
+ vec![2],
+ ),
+ _ => (
+ Predicate::negate(Predicate::and(vec![partition, data])),
+ vec![1],
+ ),
+ };
+ let messages = table
+ .new_write_builder()
+ .new_update()
+ .unwrap()
+ .delete_by_predicate(Some(predicate))
+ .await
+ .unwrap();
+ assert!(messages
+ .iter()
+ .all(|message| message.deleted_files.is_empty()));
+ commit(&table, messages).await;
+ assert_eq!(ids(&table).await, expected);
+ }
+}
+
+#[tokio::test]
+async fn data_delete_requires_dv_before_empty_or_no_match_shortcuts() {
+ for seeded in [false, true] {
+ let table = table(&[("deletion-vectors.enabled", "false")]).await;
+ if seeded {
+ seed(&table).await;
+ }
+ let pb = PredicateBuilder::new(table.schema().fields());
+ for predicate in [
+ None,
+ Some(pb.equal("id", Datum::Int(999)).unwrap()),
+ Some(Predicate::and(vec![
+ pb.equal("p", Datum::String("a".into())).unwrap(),
+ pb.equal("id", Datum::Int(999)).unwrap(),
+ ])),
+ ] {
+ let error = table
+ .new_write_builder()
+ .new_update()
+ .unwrap()
+ .delete_by_predicate(predicate)
+ .await
+ .unwrap_err();
+ assert!(
+ error.to_string().contains("deletion-vectors.enabled"),
+ "{error}"
+ );
+ }
+ }
+}
+
+#[tokio::test]
+async fn empty_table_no_match_and_delete_all() {
+ let table = table(&[]).await;
+ let update = table.new_write_builder().new_update().unwrap();
+ assert!(update.delete_by_predicate(None).await.unwrap().is_empty());
+ seed(&table).await;
+ let pb = PredicateBuilder::new(table.schema().fields());
+ for predicate in [
+ pb.equal("id", Datum::Int(999)).unwrap(),
+ pb.equal("p", Datum::String("missing".into())).unwrap(),
+ ] {
+ assert!(update
+ .delete_by_predicate(Some(predicate))
+ .await
+ .unwrap()
+ .is_empty());
+ }
+ commit(&table, update.delete_by_predicate(None).await.unwrap()).await;
+ assert!(ids(&table).await.is_empty());
+ assert!(update.delete_by_predicate(None).await.unwrap().is_empty());
+}
+
+#[tokio::test]
+async fn
partition_delete_removes_dv_and_global_indexes_and_retains_other_partition_indexes()
{
+ let table = table(&[("global-index.enabled", "true")]).await;
+ seed(&table).await;
+ table
+ .copy_with_options(HashMap::from([(
+ "deletion-vectors.enabled".into(),
+ "false".into(),
+ )]))
+ .new_global_index_build_builder()
+ .with_index_column("v")
+ .with_index_type("bitmap")
+ .execute()
+ .await
+ .unwrap();
+ let pb = PredicateBuilder::new(table.schema().fields());
+ let update = table.new_write_builder().new_update().unwrap();
+ commit(
+ &table,
+ update
+ .delete_by_predicate(Some(pb.equal("id", Datum::Int(0)).unwrap()))
+ .await
+ .unwrap(),
+ )
+ .await;
+ let before = index_names(&table).await;
+ let messages = update
+ .delete_by_predicate(Some(pb.equal("p",
Datum::String("a".into())).unwrap()))
+ .await
+ .unwrap();
+ let deleted: Vec<_> = messages
+ .iter()
+ .flat_map(|message| &message.deleted_index_files)
+ .map(|index| index.file_name.clone())
+ .collect();
+ assert_eq!(deleted.len(), 2);
+ commit(&table, messages).await;
+ let mut expected: Vec<_> = before
+ .into_iter()
+ .filter(|name| !deleted.contains(name))
+ .collect();
+ expected.sort();
+ assert_eq!(index_names(&table).await, expected);
+ assert_eq!(ids(&table).await, vec![2, 3]);
+}
+
+#[tokio::test]
+async fn
data_predicate_delete_includes_unindexed_rows_in_partial_and_detail_modes() {
+ for mode in ["PARTIAL", "DETAIL"] {
+ let table = table(&[("global-index.enabled", "true")]).await;
+ seed(&table).await;
+ table
+ .copy_with_options(HashMap::from([(
+ "deletion-vectors.enabled".into(),
+ "false".into(),
+ )]))
+ .new_global_index_build_builder()
+ .with_index_column("v")
+ .with_index_type("bitmap")
+ .execute()
+ .await
+ .unwrap();
+ write_batch(&table, &batch(vec![4], vec![Some("a")], vec![20],
vec![1])).await;
+ let table = table.copy_with_options(HashMap::from([(
+ "scalar-index.search-mode".into(),
+ mode.into(),
+ )]));
+ let pb = PredicateBuilder::new(table.schema().fields());
+ let messages = table
+ .new_write_builder()
+ .new_update()
+ .unwrap()
+ .delete_by_predicate(Some(pb.equal("v", Datum::Int(20)).unwrap()))
+ .await
+ .unwrap();
+ commit(&table, messages).await;
+ assert_eq!(ids(&table).await, vec![0, 2, 3]);
+ }
+}
+
+#[tokio::test]
+async fn partition_delete_preserves_concurrent_append_in_selected_partition() {
+ let table = table(&[("deletion-vectors.enabled", "false")]).await;
+ seed(&table).await;
+ let pb = PredicateBuilder::new(table.schema().fields());
+ let messages = table
+ .new_write_builder()
+ .new_update()
+ .unwrap()
+ .delete_by_predicate(Some(pb.equal("p",
Datum::String("a".into())).unwrap()))
+ .await
+ .unwrap();
+ write_batch(&table, &batch(vec![4], vec![Some("a")], vec![50],
vec![1])).await;
+ commit(&table, messages).await;
+ assert_eq!(ids(&table).await, vec![2, 3, 4]);
+}
+
+#[tokio::test]
+async fn row_delete_rejects_concurrent_dv_replacement() {
+ let table = table(&[]).await;
+ seed(&table).await;
+ let pb = PredicateBuilder::new(table.schema().fields());
+ let update = table.new_write_builder().new_update().unwrap();
+ let first = update
+ .delete_by_predicate(Some(pb.equal("id", Datum::Int(0)).unwrap()))
+ .await
+ .unwrap();
+ commit(
+ &table,
+ update
+ .delete_by_predicate(Some(pb.equal("id", Datum::Int(1)).unwrap()))
+ .await
+ .unwrap(),
+ )
+ .await;
+ let error = table
+ .new_write_builder()
+ .new_commit()
+ .commit(first)
+ .await
+ .unwrap_err();
+ assert!(
+ error.to_string().contains("Conflicting deletion vectors"),
+ "{error}"
+ );
+ assert_eq!(ids(&table).await, vec![0, 2, 3]);
+}
+
+#[tokio::test]
+async fn
predicate_delete_rejects_time_travel_auth_and_inconsistent_field_indices() {
+ let table = table(&[]).await;
+ seed(&table).await;
+ let pb = PredicateBuilder::new(table.schema().fields());
+ let predicate = pb.equal("p", Datum::String("a".into())).unwrap();
+ let mut wrong = predicate.clone();
+ if let Predicate::Leaf { index, .. } = &mut wrong {
+ *index = 3;
+ }
+ let error = table
+ .new_write_builder()
+ .new_update()
+ .unwrap()
+ .delete_by_predicate(Some(wrong))
+ .await
+ .unwrap_err();
+ assert!(error
+ .to_string()
+ .contains("does not match the table schema"));
+ let historical =
+ table.copy_with_options(HashMap::from([("scan.snapshot-id".into(),
"1".into())]));
+ assert!(historical.new_write_builder().new_update().is_err());
+ let auth = table.copy_with_options(HashMap::from([(
+ "query-auth.enabled".into(),
+ "true".into(),
+ )]));
+ let error = auth
+ .new_write_builder()
+ .new_update()
+ .unwrap()
+ .delete_by_predicate(Some(predicate))
+ .await
+ .unwrap_err();
+ assert!(error.to_string().contains("query-auth.enabled"));
+ assert_eq!(ids(&table).await, vec![0, 1, 2, 3]);
+}
+
+#[tokio::test]
+async fn
partition_deletes_support_append_pk_and_pending_files_without_row_tracking() {
+ for bucket in ["-1", "1", "-2"] {
+ let table = table_with_primary_keys(
+ &[
+ ("data-evolution.enabled", "false"),
+ ("row-tracking.enabled", "false"),
+ ("deletion-vectors.enabled", "false"),
+ ("bucket", bucket),
+ ],
+ bucket != "-1",
+ )
+ .await;
+ seed(&table).await;
+ let pb = PredicateBuilder::new(table.schema().fields());
+ let messages = table
+ .new_write_builder()
+ .new_update()
+ .unwrap()
+ .delete_by_predicate(Some(pb.equal("p",
Datum::String("a".into())).unwrap()))
+ .await
+ .unwrap();
+ assert_eq!(messages.len(), 1);
+ assert!(!messages[0].deleted_files.is_empty());
+ assert_eq!(messages[0].bucket, if bucket == "-2" { -2 } else { 0 });
+ commit(&table, messages).await;
+ if bucket != "-2" {
+ assert_eq!(ids(&table).await, vec![2, 3]);
+ } else {
+ let latest = table
+ .snapshot_manager()
+ .get_latest_snapshot()
+ .await
+ .unwrap()
+ .unwrap();
+ assert_eq!(latest.total_record_count(), Some(2));
+ }
+ }
+}
