This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion.git
The following commit(s) were added to refs/heads/main by this push:
new 2306a4b759 fix(core): reject a DELETE or an UPDATE whose WHERE clause
cannot reach the provider (#24657)
2306a4b759 is described below
commit 2306a4b7599dc88490c0b39f082b4cdf554a5fb9
Author: mck <[email protected]>
AuthorDate: Thu Sep 24 05:42:03 2026 +0000
fix(core): reject a DELETE or an UPDATE whose WHERE clause cannot reach the
provider (#24657)
## Which issue does this PR close?
https://github.com/apache/datafusion/issues/24654
## Rationale for this change
See ticket.
## What changes are included in this PR?
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.
## Are these changes tested?
Only with the tests provided in this patch, which are based on the
assumptions made in the ticket description.
## Are there any user-facing changes?
?
---
datafusion/core/src/physical_planner.rs | 230 +++++++++++++---
.../tests/custom_sources_cases/dml_planning.rs | 302 ++++++++++++++++++++-
datafusion/sqllogictest/test_files/dml_delete.slt | 114 ++++++++
datafusion/sqllogictest/test_files/dml_update.slt | 123 +++++++++
4 files changed, 729 insertions(+), 40 deletions(-)
diff --git a/datafusion/core/src/physical_planner.rs
b/datafusion/core/src/physical_planner.rs
index c88c34645b..24ba2fa3da 100644
--- a/datafusion/core/src/physical_planner.rs
+++ b/datafusion/core/src/physical_planner.rs
@@ -57,7 +57,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;
@@ -794,17 +794,29 @@ impl DefaultPhysicalPlanner {
target,
op: WriteOp::Delete,
input,
- ..
+ output_schema,
}) => {
if let Some(provider) =
target.downcast_ref::<DefaultTableSource>() {
- 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}'"))
- })?
+ let allowed_refs = collect_dml_target_refs(input,
table_name)?;
+ match classify_dml_input(input, table_name, &allowed_refs,
"DELETE")?
+ {
+ DmlInput::NoRows => {
+
zero_rows_affected_exec(Arc::clone(output_schema.inner()))?
+ }
+ DmlInput::Filters => {
+ let filters =
+ extract_dml_filters(input, table_name,
&allowed_refs)?;
+ 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"
@@ -816,21 +828,33 @@ impl DefaultPhysicalPlanner {
target,
op: WriteOp::Update,
input,
- ..
+ output_schema,
}) => {
if let Some(provider) =
target.downcast_ref::<DefaultTableSource>() {
- // 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}'"))
- })?
+ let allowed_refs = collect_dml_target_refs(input,
table_name)?;
+ match classify_dml_input(input, table_name, &allowed_refs,
"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,
&allowed_refs)?;
+ // 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"
@@ -2252,6 +2276,149 @@ 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,
+}
+
+/// Collect the table references that a predicate of a DELETE or an UPDATE may
+/// name: the target table itself, and the alias of every scan of the target
+/// table in the input plan.
+///
+/// Both [`classify_dml_input`] and [`extract_dml_filters`] need this set, so
the
+/// caller collects it once and passes it to each of them.
+fn collect_dml_target_refs(
+ input: &Arc<LogicalPlan>,
+ target: &TableReference,
+) -> Result<Vec<TableReference>> {
+ let mut allowed_refs = vec![target.clone()];
+ input.apply(|node| {
+ if let LogicalPlan::SubqueryAlias(alias) = node
+ // Check if this alias points to the target table
+ && 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)
+ })?;
+ Ok(allowed_refs)
+}
+
+/// 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
+/// - `allowed_refs`: the target table and its aliases, from
[`collect_dml_target_refs`]
+/// - `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
+/// * a "not implemented" error when part of the `WHERE` clause cannot reach
the provider.
+fn classify_dml_input(
+ input: &Arc<LogicalPlan>,
+ target: &TableReference,
+ allowed_refs: &[TableReference],
+ op: &str,
+) -> Result<DmlInput> {
+ 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` carries no predicate, so it reaches the provider as no
+ // filter at all and `DELETE FROM t LIMIT n` deletes every matching
+ // row. `UPDATE ... LIMIT` is already rejected by the SQL planner.
+ // That gap is separate from this one, and it is tracked
separately.
+ | 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<Schema>) -> Result<Arc<dyn
ExecutionPlan>> {
+ 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
@@ -2268,6 +2435,7 @@ fn get_physical_expr_pair(
/// # Parameters
/// - `input`: The logical plan tree to extract filters from (typically a
DELETE or UPDATE plan)
/// - `target`: The target table reference to scope filter extraction
(prevents multi-table filter leakage)
+/// - `allowed_refs`: The target table and its aliases, from
[`collect_dml_target_refs`]
///
/// # Returns
/// A vector of unqualified filter expressions that can be passed to the
TableProvider for execution.
@@ -2277,28 +2445,16 @@ fn get_physical_expr_pair(
fn extract_dml_filters(
input: &Arc<LogicalPlan>,
target: &TableReference,
+ allowed_refs: &[TableReference],
) -> Result<Vec<Expr>> {
let mut filters = Vec::new();
- let mut allowed_refs = vec![target.clone()];
-
- // First pass: collect any alias references to the target table
- input.apply(|node| {
- if let LogicalPlan::SubqueryAlias(alias) = node
- // Check if this alias points to the target table
- && 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)
- })?;
input.apply(|node| {
match node {
LogicalPlan::Filter(filter) => {
// Split AND predicates into individual expressions
for predicate in split_conjunction(&filter.predicate) {
- if predicate_is_on_target_multi(predicate, &allowed_refs)?
{
+ if predicate_is_on_target_multi(predicate, allowed_refs)? {
filters.push(predicate.clone());
}
}
diff --git a/datafusion/core/tests/custom_sources_cases/dml_planning.rs
b/datafusion/core/tests/custom_sources_cases/dml_planning.rs
index cb5b134fab..8af16cea2e 100644
--- a/datafusion/core/tests/custom_sources_cases/dml_planning.rs
+++ b/datafusion/core/tests/custom_sources_cases/dml_planning.rs
@@ -19,17 +19,21 @@
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,
provider_as_source};
use datafusion::error::Result;
use datafusion::execution::context::{SessionConfig, SessionContext};
+use datafusion::logical_expr::dml::{DmlStatement, WriteOp};
use datafusion::logical_expr::{
- Expr, LogicalPlan, TableProviderFilterPushDown, TableScan,
+ Expr, LogicalPlan, LogicalPlanBuilder, TableProviderFilterPushDown,
TableScan, col,
+ lit,
};
+use datafusion::physical_planner::{DefaultPhysicalPlanner, PhysicalPlanner};
use datafusion_catalog::Session;
-use datafusion_common::ScalarValue;
use datafusion_common::tree_node::{TreeNode, TreeNodeRecursion};
+use datafusion_common::{DataFusionError, ScalarValue};
use datafusion_physical_plan::ExecutionPlan;
use datafusion_physical_plan::empty::EmptyExec;
@@ -804,3 +808,295 @@ 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::<UInt64Array>()
+ .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<dyn TableProvider>)?;
+ 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<dyn TableProvider>)?;
+ 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<dyn TableProvider>)?;
+ 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<dyn TableProvider>)?;
+ 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<dyn TableProvider>)?;
+
+ 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<dyn TableProvider>)?;
+
+ 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(())
+}
+
+#[tokio::test]
+async fn test_dml_partial_predicate_is_rejected() -> Result<()> {
+ for sql in [
+ "DELETE FROM t WHERE value > 10 AND id IN (SELECT id FROM src)",
+ "UPDATE t SET value = 1 WHERE value > 10 AND id IN (SELECT id FROM
src)",
+ ] {
+ let delete_provider =
Arc::new(CaptureDeleteProvider::new(test_schema()));
+ let update_provider =
Arc::new(CaptureUpdateProvider::new(test_schema()));
+ let provider: Arc<dyn TableProvider> = if sql.starts_with("DELETE") {
+ delete_provider.clone()
+ } else {
+ update_provider.clone()
+ };
+ let ctx = SessionContext::new();
+ ctx.register_table("t", provider)?;
+ register_source_table(&ctx)?;
+
+ let result = ctx.sql(sql).await?.collect().await;
+ assert!(
+ delete_provider.captured_filters().is_none()
+ && update_provider.captured_filters().is_none()
+ && update_provider.captured_assignments().is_none(),
+ "a partial WHERE clause must not reach a provider: {sql}"
+ );
+ let err = result.expect_err("the subquery restriction must not be
lost");
+ assert!(
+ err.to_string().contains("IN or an EXISTS subquery"),
+ "{err}"
+ );
+ }
+ Ok(())
+}
+
+#[tokio::test]
+async fn test_delete_alias_scoping() -> Result<()> {
+ // Bypass optimization so the alias remains in the input to the physical
planner.
+ for scan_name in ["t", "src"] {
+ let provider = Arc::new(CaptureDeleteProvider::new(test_schema()));
+ let target = provider_as_source(provider.clone());
+ let source = if scan_name == "t" {
+ Arc::clone(&target)
+ } else {
+ provider_as_source(Arc::new(MemTable::try_new(test_schema(),
vec![vec![]])?))
+ };
+ let input = LogicalPlanBuilder::scan(scan_name, source, None)?
+ .alias("a")?
+ .filter(col("a.id").eq(lit(1)))?
+ .build()?;
+ let plan = LogicalPlan::Dml(DmlStatement::new(
+ "t".into(),
+ target,
+ WriteOp::Delete,
+ Arc::new(input),
+ ));
+ let result = DefaultPhysicalPlanner::default()
+ .create_physical_plan(&plan, &SessionContext::new().state())
+ .await;
+
+ if scan_name == "t" {
+ result?;
+ assert_eq!(
+ provider.captured_filters(),
+ Some(vec![col("id").eq(lit(1))]),
+ "the target alias must be accepted and its qualifier stripped"
+ );
+ } else {
+ assert!(
+ provider.captured_filters().is_none(),
+ "a foreign predicate must not be dropped before calling
delete_from()"
+ );
+ let err =
+ result.expect_err("an alias of another table is not a target
alias");
+ assert!(matches!(err, DataFusionError::NotImplemented(_)),
"{err}");
+ assert!(
+ err.to_string().contains("references another table"),
+ "{err}"
+ );
+ }
+ }
+ Ok(())
+}
+
+#[tokio::test]
+async fn test_delete_aggregate_input_is_rejected() -> Result<()> {
+ let provider = Arc::new(CaptureDeleteProvider::new(test_schema()));
+ let target = provider_as_source(provider.clone());
+ let input = LogicalPlanBuilder::scan("t", Arc::clone(&target), None)?
+ .aggregate(
+ vec![col("id"), col("status"), col("value")],
+ Vec::<Expr>::new(),
+ )?
+ .build()?;
+ let plan = LogicalPlan::Dml(DmlStatement::new(
+ "t".into(),
+ target,
+ WriteOp::Delete,
+ Arc::new(input),
+ ));
+ let result = DefaultPhysicalPlanner::default()
+ .create_physical_plan(&plan, &SessionContext::new().state())
+ .await;
+
+ assert!(
+ provider.captured_filters().is_none(),
+ "aggregation must not become an unrestricted delete_from() call"
+ );
+ let err = result.expect_err("aggregation cannot be represented by provider
filters");
+ assert!(matches!(err, DataFusionError::NotImplemented(_)), "{err}");
+ assert!(err.to_string().contains("Aggregate"), "{err}");
+ Ok(())
+}
diff --git a/datafusion/sqllogictest/test_files/dml_delete.slt
b/datafusion/sqllogictest/test_files/dml_delete.slt
index 296baa729f..dc68b8c56d 100644
--- a/datafusion/sqllogictest/test_files/dml_delete.slt
+++ b/datafusion/sqllogictest/test_files/dml_delete.slt
@@ -247,3 +247,117 @@ SELECT * FROM test_delete_error;
statement ok
DROP TABLE test_delete_error;
+
+# 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);
+
+# The ordinary predicate matches rows 2 and 3, but the full condition only
matches 2.
+# Passing only the ordinary predicate to the provider would delete too many
rows.
+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 > 1 AND column1 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;
+
+# Test the same statements with the optimizer switched off
+# With `max_passes = 0` no rule rewrites the subquery into a semi join and no
+# rule folds an always-false predicate, so the whole WHERE clause reaches the
+# table provider. A subquery predicate is rejected there, when the provider
+# compiles it to a physical expression, so no path deletes every row.
+statement ok
+set datafusion.optimizer.max_passes = 0;
+
+statement ok
+CREATE TABLE test_delete_unopt AS VALUES (1), (2), (3);
+
+statement ok
+CREATE TABLE test_delete_unopt_src AS VALUES (2);
+
+statement error DataFusion error: This feature is not implemented: Physical
plan does not support logical expression InSubquery
+DELETE FROM test_delete_unopt WHERE column1 IN (SELECT column1 FROM
test_delete_unopt_src);
+
+statement error DataFusion error: This feature is not implemented: Physical
plan does not support logical expression Exists
+DELETE FROM test_delete_unopt WHERE EXISTS (SELECT 1 FROM
test_delete_unopt_src WHERE test_delete_unopt_src.column1 =
test_delete_unopt.column1);
+
+statement error DataFusion error: This feature is not implemented: Physical
plan does not support logical expression InSubquery
+DELETE FROM test_delete_unopt WHERE column1 NOT IN (SELECT column1 FROM
test_delete_unopt_src);
+
+# The always-false predicate reaches the provider, which matches no row with it
+query I
+DELETE FROM test_delete_unopt WHERE false;
+----
+0
+
+query I
+DELETE FROM test_delete_unopt WHERE 1 = 2;
+----
+0
+
+query I rowsort
+SELECT * FROM test_delete_unopt;
+----
+1
+2
+3
+
+statement ok
+DROP TABLE test_delete_unopt_src;
+
+statement ok
+DROP TABLE test_delete_unopt;
+
+statement ok
+RESET datafusion.optimizer.max_passes;
diff --git a/datafusion/sqllogictest/test_files/dml_update.slt
b/datafusion/sqllogictest/test_files/dml_update.slt
index d0712afff2..f10185a396 100644
--- a/datafusion/sqllogictest/test_files/dml_update.slt
+++ b/datafusion/sqllogictest/test_files/dml_update.slt
@@ -365,3 +365,126 @@ SELECT * FROM test_update_not_null;
statement ok
DROP TABLE test_update_not_null;
+
+# 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);
+
+# The ordinary predicate matches rows 2 and 3, but the full condition only
matches 2.
+# Passing only the ordinary predicate to the provider would update too many
rows.
+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 > 1 AND id IN (SELECT id
FROM test_update_subquery_src);
+
+# 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;
+
+# Test the same statements with the optimizer switched off
+# With `max_passes = 0` no rule rewrites the subquery into a semi join and no
+# rule folds an always-false predicate, so the whole WHERE clause reaches the
+# table provider. A subquery predicate is rejected there, when the provider
+# compiles it to a physical expression, so no path updates every row.
+statement ok
+set datafusion.optimizer.max_passes = 0;
+
+statement ok
+CREATE TABLE test_update_unopt(id INT, name VARCHAR);
+
+statement ok
+INSERT INTO test_update_unopt VALUES (1, 'a'), (2, 'b'), (3, 'c');
+
+statement ok
+CREATE TABLE test_update_unopt_src(id INT);
+
+statement ok
+INSERT INTO test_update_unopt_src VALUES (2);
+
+statement error DataFusion error: This feature is not implemented: Physical
plan does not support logical expression InSubquery
+UPDATE test_update_unopt SET name = 'z' WHERE id IN (SELECT id FROM
test_update_unopt_src);
+
+statement error DataFusion error: This feature is not implemented: Physical
plan does not support logical expression Exists
+UPDATE test_update_unopt SET name = 'z' WHERE EXISTS (SELECT 1 FROM
test_update_unopt_src WHERE test_update_unopt_src.id = test_update_unopt.id);
+
+# The always-false predicate reaches the provider, which matches no row with it
+query I
+UPDATE test_update_unopt SET name = 'z' WHERE false;
+----
+0
+
+query I
+UPDATE test_update_unopt SET name = 'z' WHERE 1 = 2;
+----
+0
+
+query IT rowsort
+SELECT * FROM test_update_unopt;
+----
+1 a
+2 b
+3 c
+
+statement ok
+DROP TABLE test_update_unopt_src;
+
+statement ok
+DROP TABLE test_update_unopt;
+
+statement ok
+RESET datafusion.optimizer.max_passes;
\ No newline at end of file
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
