From 5f3e301801e3c27f3a3dd69974f71d97e811faad Mon Sep 17 00:00:00 2001 From: mck Date: Tue, 25 Aug 2026 10:10:42 +0200 Subject: [PATCH] fix(core): reject a DELETE or an UPDATE whose WHERE clause cannot reach the provider A `DELETE` or an `UPDATE` whose `WHERE` clause holds an `IN` or an `EXISTS` subquery changed every row of the target table, and reported the whole table as affected. The optimizer rewrites the subquery into a semi join, so the condition leaves the `Filter` nodes that `extract_dml_filters()` reads. The provider then received an empty filter list, which is the encoding for "no WHERE clause", and applied the statement to all rows. An always-false `WHERE` clause reached the provider the same way. The simplifier folds the predicate into an empty relation, so again no filter survived, and a `DELETE FROM t WHERE false` emptied the table. Add `classify_dml_input()`, which walks the input plan of a `DELETE` or an `UPDATE` before the provider hook runs: - an empty relation means that no row matches, so the statement reports a count of 0 and the hook is not called; - a join, a predicate on another table, or any other node that restricts or multiplies rows raises a "not implemented" error, and the hook is not called. The hook stays untouched in every rejected case, so a provider that writes to durable storage cannot lose rows. Co-Authored-By: Claude Opus 5 --- datafusion/core/src/physical_planner.rs | 190 +++++++++++++++--- .../custom_sources_cases/dml_planning.rs | 183 ++++++++++++++++- .../sqllogictest/test_files/dml_delete.slt | 59 ++++++ .../sqllogictest/test_files/dml_update.slt | 65 ++++++ 4 files changed, 473 insertions(+), 24 deletions(-) diff --git a/datafusion/core/src/physical_planner.rs b/datafusion/core/src/physical_planner.rs index 2792c9c7a6faa..3f6ae287c481e 100644 --- a/datafusion/core/src/physical_planner.rs +++ b/datafusion/core/src/physical_planner.rs @@ -56,7 +56,7 @@ use crate::physical_plan::{ }; use crate::schema_equivalence::schema_satisfied_by; -use arrow::array::{RecordBatch, builder::StringBuilder}; +use arrow::array::{ArrayRef, RecordBatch, UInt64Array, builder::StringBuilder}; use arrow::compute::SortOptions; use arrow::datatypes::Schema; use arrow_schema::Field; @@ -793,17 +793,26 @@ impl DefaultPhysicalPlanner { target, op: WriteOp::Delete, input, - .. + output_schema, }) => { if let Some(provider) = target.downcast_ref::() { - let filters = extract_dml_filters(input, table_name)?; - provider - .table_provider - .delete_from(session_state, filters) - .await - .map_err(|e| { - e.context(format!("DELETE operation on table '{table_name}'")) - })? + match classify_dml_input(input, table_name, "DELETE")? { + DmlInput::NoRows => { + zero_rows_affected_exec(Arc::clone(output_schema.inner()))? + } + DmlInput::Filters => { + let filters = extract_dml_filters(input, table_name)?; + provider + .table_provider + .delete_from(session_state, filters) + .await + .map_err(|e| { + e.context(format!( + "DELETE operation on table '{table_name}'" + )) + })? + } + } } else { return exec_err!( "Table source can't be downcasted to DefaultTableSource" @@ -815,21 +824,30 @@ impl DefaultPhysicalPlanner { target, op: WriteOp::Update, input, - .. + output_schema, }) => { if let Some(provider) = target.downcast_ref::() { - // For UPDATE, the assignments are encoded in the projection of input - // We pass the filters and let the provider handle the projection - let filters = extract_dml_filters(input, table_name)?; - // Extract assignments from the projection in input plan - let assignments = extract_update_assignments(input)?; - provider - .table_provider - .update(session_state, assignments, filters) - .await - .map_err(|e| { - e.context(format!("UPDATE operation on table '{table_name}'")) - })? + match classify_dml_input(input, table_name, "UPDATE")? { + DmlInput::NoRows => { + zero_rows_affected_exec(Arc::clone(output_schema.inner()))? + } + DmlInput::Filters => { + // For UPDATE, the assignments are encoded in the projection of input + // We pass the filters and let the provider handle the projection + let filters = extract_dml_filters(input, table_name)?; + // Extract assignments from the projection in input plan + let assignments = extract_update_assignments(input)?; + provider + .table_provider + .update(session_state, assignments, filters) + .await + .map_err(|e| { + e.context(format!( + "UPDATE operation on table '{table_name}'" + )) + })? + } + } } else { return exec_err!( "Table source can't be downcasted to DefaultTableSource" @@ -2202,6 +2220,132 @@ fn get_physical_expr_pair( Ok((physical_expr, physical_name)) } +/// How a DELETE or an UPDATE reaches its target table. +/// +/// The `filters` argument of [`TableProvider::delete_from`] and +/// [`TableProvider::update`] is the only channel that carries the `WHERE` clause +/// to the provider, and an empty vector means "no `WHERE` clause, so every row". +/// A plan whose row restriction cannot travel through that channel must +/// therefore never reach the provider. +/// +/// [`TableProvider::delete_from`]: datafusion_catalog::TableProvider::delete_from +/// [`TableProvider::update`]: datafusion_catalog::TableProvider::update +enum DmlInput { + /// Every row restriction of the statement reaches the provider as a filter. + Filters, + /// No row matches, so the statement affects no rows and the provider is not + /// called at all. + NoRows, +} + +/// Check that the input plan of a DELETE or an UPDATE can reach the table +/// provider without losing part of its `WHERE` clause. +/// +/// The optimizer rewrites an `IN` or an `EXISTS` subquery into a semi join, and +/// it folds an always-false predicate into an empty relation. In both cases the +/// condition leaves the `Filter` nodes that [`extract_dml_filters`] reads, and +/// the provider would see an empty filter list and change every row. +/// +/// # Parameters +/// - `input`: the input plan of the DELETE or the UPDATE +/// - `target`: the target table of the statement +/// - `op`: `"DELETE"` or `"UPDATE"`, used in the error message +/// +/// # Returns +/// [`DmlInput::Filters`] when the provider may be called, [`DmlInput::NoRows`] +/// when the statement matches no row, and a "not implemented" error when part of +/// the `WHERE` clause cannot reach the provider. +fn classify_dml_input( + input: &Arc, + target: &TableReference, + op: &str, +) -> Result { + let mut allowed_refs = vec![target.clone()]; + input.apply(|node| { + if let LogicalPlan::SubqueryAlias(alias) = node + && let LogicalPlan::TableScan(scan) = alias.input.as_ref() + && scan.table_name.resolved_eq(target) + { + allowed_refs.push(TableReference::bare(alias.alias.to_string())); + } + Ok(TreeNodeRecursion::Continue) + })?; + + let mut result = DmlInput::Filters; + input.apply(|node| { + match node { + // An empty relation means the optimizer proved that no row matches, + // so the statement affects no rows. + LogicalPlan::EmptyRelation(empty) if !empty.produce_one_row => { + result = DmlInput::NoRows; + return Ok(TreeNodeRecursion::Stop); + } + // A join carries the condition in its `on` clause, where + // `extract_dml_filters` cannot read it. The optimizer builds one for + // an `IN` or an `EXISTS` subquery. + LogicalPlan::Join(join) => { + return not_impl_err!( + "{op} on table '{target}' with an IN or an EXISTS subquery in its \ + WHERE clause is not supported: the optimizer rewrites the subquery \ + into a {} join, and the condition does not reach the table provider", + join.join_type + ); + } + LogicalPlan::Filter(filter) => { + // A predicate on another table restricts the rows of the target + // table, and the provider cannot evaluate it. + for predicate in split_conjunction(&filter.predicate) { + if !predicate_is_on_target_multi(predicate, &allowed_refs)? { + return not_impl_err!( + "{op} on table '{target}' with a WHERE clause that \ + references another table is not supported" + ); + } + } + } + // Plans that pass every row of the target table through, or that + // hold no row restriction of their own. + LogicalPlan::TableScan(_) + | LogicalPlan::Projection(_) + | LogicalPlan::SubqueryAlias(_) + | LogicalPlan::Sort(_) + | LogicalPlan::Repartition(_) + // A `Limit` reaches the provider as no filter at all, so a DELETE + // ignores it. That is a separate gap, kept as it is here. + | LogicalPlan::Limit(_) + // A subquery expression that survives to this point fails later, + // when the provider compiles the filter it belongs to. + | LogicalPlan::Subquery(_) => {} + // Everything else either restricts or multiplies the rows of the + // target table in a way that no filter list can express. + other => { + return not_impl_err!( + "{op} on table '{target}' is not supported: the statement plan \ + contains \"{}\", and its effect on the rows cannot reach the table \ + provider as a filter", + other.display() + ); + } + } + Ok(TreeNodeRecursion::Continue) + })?; + + Ok(result) +} + +/// Build a plan that reports no rows affected, for a DELETE or an UPDATE that +/// matches no row. `schema` is the output schema of the statement, one `count` +/// column of type `UInt64`. +fn zero_rows_affected_exec(schema: Arc) -> Result> { + let count = Arc::new(UInt64Array::from(vec![0_u64])) as ArrayRef; + let batch = RecordBatch::try_new(Arc::clone(&schema), vec![count])?; + Ok(MemorySourceConfig::try_new_exec( + &[vec![batch]], + schema, + None, + )?) +} + /// Extract filter predicates from a DML input plan (DELETE/UPDATE). /// /// Walks the logical plan tree and collects Filter predicates and any filters diff --git a/datafusion/core/tests/custom_sources_cases/dml_planning.rs b/datafusion/core/tests/custom_sources_cases/dml_planning.rs index 6ae942bfbb88b..039062a8e512f 100644 --- a/datafusion/core/tests/custom_sources_cases/dml_planning.rs +++ b/datafusion/core/tests/custom_sources_cases/dml_planning.rs @@ -19,9 +19,10 @@ use std::sync::{Arc, Mutex}; +use arrow::array::{Int32Array, RecordBatch, UInt64Array}; use arrow::datatypes::{DataType, Field, Schema, SchemaRef}; use async_trait::async_trait; -use datafusion::datasource::{TableProvider, TableType}; +use datafusion::datasource::{MemTable, TableProvider, TableType}; use datafusion::error::Result; use datafusion::execution::context::{SessionConfig, SessionContext}; use datafusion::logical_expr::{ @@ -804,3 +805,183 @@ async fn test_unsupported_table_truncate() -> Result<()> { Ok(()) } + +/// Register a source table named `src` with one row, for the subquery of a +/// DELETE or an UPDATE. The table holds a row so that the optimizer keeps the +/// semi join instead of folding it into an empty relation. +fn register_source_table(ctx: &SessionContext) -> Result<()> { + let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Int32, false)])); + let batch = RecordBatch::try_new( + Arc::clone(&schema), + vec![Arc::new(Int32Array::from(vec![1]))], + )?; + let source = MemTable::try_new(schema, vec![vec![batch]])?; + ctx.register_table("src", Arc::new(source))?; + Ok(()) +} + +/// Read the single `count` value of a DML result. +fn rows_affected(batches: &[RecordBatch]) -> u64 { + assert_eq!(batches.len(), 1, "a DML statement returns one batch"); + let counts = batches[0] + .column(0) + .as_any() + .downcast_ref::() + .expect("the count column is UInt64"); + assert_eq!(counts.len(), 1, "a DML statement returns one row"); + counts.value(0) +} + +/// A DELETE whose WHERE clause holds an IN subquery must fail, and it must not +/// call the provider. The optimizer rewrites the subquery into a LeftSemi join, +/// so no filter reaches the provider, and a provider that reads an empty filter +/// list as "no WHERE clause" would delete every row. +#[tokio::test] +async fn test_delete_in_subquery_is_rejected() -> Result<()> { + let provider = Arc::new(CaptureDeleteProvider::new(test_schema())); + let ctx = SessionContext::new(); + ctx.register_table("t", Arc::clone(&provider) as Arc)?; + register_source_table(&ctx)?; + + let result = ctx + .sql("DELETE FROM t WHERE id IN (SELECT id FROM src)") + .await? + .collect() + .await; + + let err = result.expect_err("DELETE with an IN subquery should fail"); + assert!( + err.to_string().contains("IN or an EXISTS subquery"), + "unexpected error: {err}" + ); + assert!( + provider.captured_filters().is_none(), + "delete_from() must not be called, or the provider deletes every row" + ); + Ok(()) +} + +/// A DELETE whose WHERE clause holds a correlated EXISTS subquery must fail, +/// for the same reason as the IN subquery. +#[tokio::test] +async fn test_delete_exists_subquery_is_rejected() -> Result<()> { + let provider = Arc::new(CaptureDeleteProvider::new(test_schema())); + let ctx = SessionContext::new(); + ctx.register_table("t", Arc::clone(&provider) as Arc)?; + register_source_table(&ctx)?; + + let result = ctx + .sql("DELETE FROM t WHERE EXISTS (SELECT 1 FROM src WHERE src.id = t.id)") + .await? + .collect() + .await; + + let err = result.expect_err("DELETE with an EXISTS subquery should fail"); + assert!( + err.to_string().contains("IN or an EXISTS subquery"), + "unexpected error: {err}" + ); + assert!( + provider.captured_filters().is_none(), + "delete_from() must not be called, or the provider deletes every row" + ); + Ok(()) +} + +/// A negated subquery becomes a LeftAnti join, and must fail as well. +#[tokio::test] +async fn test_delete_not_in_subquery_is_rejected() -> Result<()> { + let provider = Arc::new(CaptureDeleteProvider::new(test_schema())); + let ctx = SessionContext::new(); + ctx.register_table("t", Arc::clone(&provider) as Arc)?; + register_source_table(&ctx)?; + + let result = ctx + .sql("DELETE FROM t WHERE id NOT IN (SELECT id FROM src)") + .await? + .collect() + .await; + + let err = result.expect_err("DELETE with a NOT IN subquery should fail"); + assert!( + err.to_string().contains("LeftAnti join"), + "unexpected error: {err}" + ); + assert!( + provider.captured_filters().is_none(), + "delete_from() must not be called, or the provider deletes every row" + ); + Ok(()) +} + +/// An UPDATE whose WHERE clause holds an IN subquery must fail, and it must not +/// call the provider. +#[tokio::test] +async fn test_update_in_subquery_is_rejected() -> Result<()> { + let provider = Arc::new(CaptureUpdateProvider::new(test_schema())); + let ctx = SessionContext::new(); + ctx.register_table("t", Arc::clone(&provider) as Arc)?; + register_source_table(&ctx)?; + + let result = ctx + .sql("UPDATE t SET value = 1 WHERE id IN (SELECT id FROM src)") + .await? + .collect() + .await; + + let err = result.expect_err("UPDATE with an IN subquery should fail"); + assert!( + err.to_string().contains("IN or an EXISTS subquery"), + "unexpected error: {err}" + ); + assert!( + provider.captured_filters().is_none(), + "update() must not be called, or the provider changes every row" + ); + assert!(provider.captured_assignments().is_none()); + Ok(()) +} + +/// A DELETE whose WHERE clause is always false affects no rows. The optimizer +/// folds the predicate into an empty relation, so no filter reaches the +/// provider, and the provider must not be called at all. +#[tokio::test] +async fn test_delete_always_false_predicate_affects_no_rows() -> Result<()> { + let provider = Arc::new(CaptureDeleteProvider::new(test_schema())); + let ctx = SessionContext::new(); + ctx.register_table("t", Arc::clone(&provider) as Arc)?; + + let batches = ctx + .sql("DELETE FROM t WHERE 1 = 2") + .await? + .collect() + .await?; + + assert_eq!(rows_affected(&batches), 0); + assert!( + provider.captured_filters().is_none(), + "delete_from() must not be called, or the provider deletes every row" + ); + Ok(()) +} + +/// An UPDATE whose WHERE clause is always false affects no rows. +#[tokio::test] +async fn test_update_always_false_predicate_affects_no_rows() -> Result<()> { + let provider = Arc::new(CaptureUpdateProvider::new(test_schema())); + let ctx = SessionContext::new(); + ctx.register_table("t", Arc::clone(&provider) as Arc)?; + + let batches = ctx + .sql("UPDATE t SET value = 1 WHERE false") + .await? + .collect() + .await?; + + assert_eq!(rows_affected(&batches), 0); + assert!( + provider.captured_filters().is_none(), + "update() must not be called, or the provider changes every row" + ); + Ok(()) +} diff --git a/datafusion/sqllogictest/test_files/dml_delete.slt b/datafusion/sqllogictest/test_files/dml_delete.slt index 3dae431ada377..81fb5b0976ffe 100644 --- a/datafusion/sqllogictest/test_files/dml_delete.slt +++ b/datafusion/sqllogictest/test_files/dml_delete.slt @@ -200,3 +200,62 @@ SELECT * FROM test_delete_in; statement ok DROP TABLE test_delete_in; + +# Test DELETE with an IN or an EXISTS subquery in the WHERE clause +# The optimizer rewrites the subquery into a semi join, so the condition cannot +# reach the table provider as a filter. DataFusion rejects the statement instead +# of deleting every row. +statement ok +CREATE TABLE test_delete_subquery AS VALUES (1), (2), (3); + +statement ok +CREATE TABLE test_delete_subquery_src AS VALUES (2); + +statement error DataFusion error: This feature is not implemented: DELETE on table 'test_delete_subquery' with an IN or an EXISTS subquery in its WHERE clause is not supported +DELETE FROM test_delete_subquery WHERE column1 IN (SELECT column1 FROM test_delete_subquery_src); + +statement error DataFusion error: This feature is not implemented: DELETE on table 'test_delete_subquery' with an IN or an EXISTS subquery in its WHERE clause is not supported +DELETE FROM test_delete_subquery WHERE EXISTS (SELECT 1 FROM test_delete_subquery_src WHERE test_delete_subquery_src.column1 = test_delete_subquery.column1); + +statement error DataFusion error: This feature is not implemented: DELETE on table 'test_delete_subquery' with an IN or an EXISTS subquery in its WHERE clause is not supported +DELETE FROM test_delete_subquery WHERE column1 NOT IN (SELECT column1 FROM test_delete_subquery_src); + +# Every row survives each rejected statement +query I rowsort +SELECT * FROM test_delete_subquery; +---- +1 +2 +3 + +statement ok +DROP TABLE test_delete_subquery_src; + +statement ok +DROP TABLE test_delete_subquery; + +# Test DELETE with an always-false WHERE clause +# The optimizer folds the predicate into an empty relation, so no filter reaches +# the table provider. The statement affects no rows. +statement ok +CREATE TABLE test_delete_false AS VALUES (1), (2), (3); + +query I +DELETE FROM test_delete_false WHERE false; +---- +0 + +query I +DELETE FROM test_delete_false WHERE 1 = 2; +---- +0 + +query I rowsort +SELECT * FROM test_delete_false; +---- +1 +2 +3 + +statement ok +DROP TABLE test_delete_false; diff --git a/datafusion/sqllogictest/test_files/dml_update.slt b/datafusion/sqllogictest/test_files/dml_update.slt index 10f74ae3970da..ca38798c77bbf 100644 --- a/datafusion/sqllogictest/test_files/dml_update.slt +++ b/datafusion/sqllogictest/test_files/dml_update.slt @@ -284,3 +284,68 @@ SELECT * FROM test_update_div; statement ok DROP TABLE test_update_div; + +# Test UPDATE with an IN or an EXISTS subquery in the WHERE clause +# The optimizer rewrites the subquery into a semi join, so the condition cannot +# reach the table provider as a filter. DataFusion rejects the statement instead +# of updating every row. +statement ok +CREATE TABLE test_update_subquery(id INT, name VARCHAR); + +statement ok +INSERT INTO test_update_subquery VALUES (1, 'a'), (2, 'b'), (3, 'c'); + +statement ok +CREATE TABLE test_update_subquery_src(id INT); + +statement ok +INSERT INTO test_update_subquery_src VALUES (2); + +statement error DataFusion error: This feature is not implemented: UPDATE on table 'test_update_subquery' with an IN or an EXISTS subquery in its WHERE clause is not supported +UPDATE test_update_subquery SET name = 'z' WHERE id IN (SELECT id FROM test_update_subquery_src); + +statement error DataFusion error: This feature is not implemented: UPDATE on table 'test_update_subquery' with an IN or an EXISTS subquery in its WHERE clause is not supported +UPDATE test_update_subquery SET name = 'z' WHERE EXISTS (SELECT 1 FROM test_update_subquery_src WHERE test_update_subquery_src.id = test_update_subquery.id); + +# Every row keeps its value after each rejected statement +query IT rowsort +SELECT * FROM test_update_subquery; +---- +1 a +2 b +3 c + +statement ok +DROP TABLE test_update_subquery_src; + +statement ok +DROP TABLE test_update_subquery; + +# Test UPDATE with an always-false WHERE clause +# The optimizer folds the predicate into an empty relation, so no filter reaches +# the table provider. The statement affects no rows. +statement ok +CREATE TABLE test_update_false(id INT, name VARCHAR); + +statement ok +INSERT INTO test_update_false VALUES (1, 'a'), (2, 'b'), (3, 'c'); + +query I +UPDATE test_update_false SET name = 'z' WHERE false; +---- +0 + +query I +UPDATE test_update_false SET name = 'z' WHERE 1 = 2; +---- +0 + +query IT rowsort +SELECT * FROM test_update_false; +---- +1 a +2 b +3 c + +statement ok +DROP TABLE test_update_false;