This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-24657-5b8fcf1b8f8b931b2e25730ffdffa3a4bc585bfe in repository https://gitbox.apache.org/repos/asf/datafusion.git
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]
