You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/04/14 03:12:23 UTC
(datafusion) branch gh-readonly-queue/main/pr-20977-f1c643a8d07ddd56461823ba8b150bd5e2064dc9 created (now e0fd16ee3f)
This is an automated email from the ASF dual-hosted git repository.
github-bot pushed a change to branch gh-readonly-queue/main/pr-20977-f1c643a8d07ddd56461823ba8b150bd5e2064dc9
in repository https://gitbox.apache.org/repos/asf/datafusion.git
at e0fd16ee3f feat(substrait): support Placeholder <-> DynamicParameter in Substrait producer/consumer (#20977)
This branch includes the following new commits:
new e0fd16ee3f feat(substrait): support Placeholder <-> DynamicParameter in Substrait producer/consumer (#20977)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion) 01/01: feat(substrait): support Placeholder <-> DynamicParameter in Substrait producer/consumer (#20977)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-bot pushed a commit to branch gh-readonly-queue/main/pr-20977-f1c643a8d07ddd56461823ba8b150bd5e2064dc9
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit e0fd16ee3fc88e58f8c3e2f63a43ccf3d4ab2f9e
Author: Bruno Volpato <br...@gmail.com>
AuthorDate: Mon Apr 13 23:12:02 2026 -0400
feat(substrait): support Placeholder <-> DynamicParameter in Substrait producer/consumer (#20977)
## Which issue does this PR close?
- Closes https://github.com/apache/datafusion/issues/20971.
## Rationale for this change
The Substrait producer previously returned `not_impl_err!` for
`Expr::Placeholder`, meaning any query plan containing parameterized
expressions (e.g. `$1`, `$2`) could not be serialized to Substrait. This
gap prevented roundtripping prepared statement plans through the
Substrait layer.
## What changes are included in this PR?
**Producer:**
- New `placeholder.rs` module with `from_placeholder` that converts
`Expr::Placeholder` to `DynamicParameter`, mapping one-based `$N`
DataFusion ids to zero-based Substrait `parameter_reference` values with
optional type information.
- `handle_placeholder` added to the `SubstraitProducer` trait.
**Consumer:**
- `consume_dynamic_parameter` converts Substrait `DynamicParameter` back
to `Expr::Placeholder`, reversing the index mapping and type conversion.
## Are these changes tested?
Yes. Five new integration tests and one unit test:
- `roundtrip_placeholder_sql_filter` — SQL-based, `WHERE a > $1`
- `roundtrip_placeholder_sql_projection` — SQL-based, `$1` in SELECT +
`$2` in WHERE
- `roundtrip_placeholder_typed_int64` — typed Int64 placeholder with
proto-level `DynamicParameter` verification
- `roundtrip_placeholder_multiple_typed` — two typed placeholders (Int64
+ Decimal128)
- `roundtrip_placeholder_typed_utf8` — Utf8 typed placeholder
- `test_parse_placeholder_index` — unit test for index parsing edge
cases
All 190 integration tests, 28 unit tests, and 3 doc-tests pass. Clippy
and fmt are clean.
## Are there any user-facing changes?
Query plans containing `Placeholder` expressions can now be serialized
to and deserialized from Substrait format. Previously this would return
an error.
Co-authored-by: Jonah Gao <jo...@msn.com>
---
.../logical_plan/consumer/substrait_consumer.rs | 16 ++-
.../src/logical_plan/producer/expr/mod.rs | 4 +-
.../src/logical_plan/producer/expr/placeholder.rs | 68 +++++++++++
.../logical_plan/producer/substrait_producer.rs | 18 ++-
.../tests/cases/roundtrip_logical_plan.rs | 134 +++++++++++++++++++++
5 files changed, 232 insertions(+), 8 deletions(-)
diff --git a/datafusion/substrait/src/logical_plan/consumer/substrait_consumer.rs b/datafusion/substrait/src/logical_plan/consumer/substrait_consumer.rs
index 93f0a494d8..65bc53ce08 100644
--- a/datafusion/substrait/src/logical_plan/consumer/substrait_consumer.rs
+++ b/datafusion/substrait/src/logical_plan/consumer/substrait_consumer.rs
@@ -366,10 +366,22 @@ pub trait SubstraitConsumer: Send + Sync + Sized {
async fn consume_dynamic_parameter(
&self,
- _expr: &DynamicParameter,
+ expr: &DynamicParameter,
_input_schema: &DFSchema,
) -> datafusion::common::Result<Expr> {
- not_impl_err!("Dynamic Parameter expression not supported")
+ let id = format!("${}", expr.parameter_reference + 1);
+ let field = expr
+ .r#type
+ .as_ref()
+ .map(|t| {
+ super::from_substrait_type_without_names(self, t).map(|dt| {
+ Arc::new(datafusion::arrow::datatypes::Field::new(&id, dt, true))
+ })
+ })
+ .transpose()?;
+ Ok(Expr::Placeholder(
+ datafusion::logical_expr::expr::Placeholder::new_with_field(id, field),
+ ))
}
// Outer Schema Stack
diff --git a/datafusion/substrait/src/logical_plan/producer/expr/mod.rs b/datafusion/substrait/src/logical_plan/producer/expr/mod.rs
index d130961596..b07a0206c5 100644
--- a/datafusion/substrait/src/logical_plan/producer/expr/mod.rs
+++ b/datafusion/substrait/src/logical_plan/producer/expr/mod.rs
@@ -20,6 +20,7 @@ mod cast;
mod field_reference;
mod if_then;
mod literal;
+mod placeholder;
mod scalar_function;
mod singular_or_list;
mod subquery;
@@ -30,6 +31,7 @@ pub use cast::*;
pub use field_reference::*;
pub use if_then::*;
pub use literal::*;
+pub use placeholder::*;
pub use scalar_function::*;
pub use singular_or_list::*;
pub use subquery::*;
@@ -142,7 +144,7 @@ pub fn to_substrait_rex(
#[expect(deprecated)]
Expr::Wildcard { .. } => not_impl_err!("Cannot convert {expr:?} to Substrait"),
Expr::GroupingSet(expr) => not_impl_err!("Cannot convert {expr:?} to Substrait"),
- Expr::Placeholder(expr) => not_impl_err!("Cannot convert {expr:?} to Substrait"),
+ Expr::Placeholder(expr) => producer.handle_placeholder(expr, schema),
Expr::OuterReferenceColumn(_, _) => {
// OuterReferenceColumn requires tracking outer query schema context for correlated
// subqueries. This is a complex feature that is not yet implemented.
diff --git a/datafusion/substrait/src/logical_plan/producer/expr/placeholder.rs b/datafusion/substrait/src/logical_plan/producer/expr/placeholder.rs
new file mode 100644
index 0000000000..44721abc0c
--- /dev/null
+++ b/datafusion/substrait/src/logical_plan/producer/expr/placeholder.rs
@@ -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.
+
+use crate::logical_plan::producer::{SubstraitProducer, to_substrait_type};
+use datafusion::common::substrait_err;
+use datafusion::logical_expr::expr::Placeholder;
+use substrait::proto::expression::RexType;
+use substrait::proto::{DynamicParameter, Expression};
+
+pub fn from_placeholder(
+ producer: &mut impl SubstraitProducer,
+ placeholder: &Placeholder,
+) -> datafusion::common::Result<Expression> {
+ let parameter_reference = parse_placeholder_index(&placeholder.id)?;
+
+ let r#type = placeholder
+ .field
+ .as_ref()
+ .map(|field| to_substrait_type(producer, field.data_type(), field.is_nullable()))
+ .transpose()?;
+
+ Ok(Expression {
+ rex_type: Some(RexType::DynamicParameter(DynamicParameter {
+ r#type,
+ parameter_reference,
+ })),
+ })
+}
+
+/// Converts a placeholder id like "$1" into a zero-based parameter index.
+/// Substrait uses zero-based `parameter_reference` while DataFusion uses
+/// one-based `$N` placeholder ids.
+fn parse_placeholder_index(id: &str) -> datafusion::common::Result<u32> {
+ let num_str = id.strip_prefix('$').unwrap_or(id);
+ match num_str.parse::<u32>() {
+ Ok(n) if n > 0 => Ok(n - 1),
+ Ok(_) => substrait_err!("Placeholder index must be >= 1, got: {id}"),
+ Err(_) => substrait_err!("Cannot parse placeholder id as numeric index: {id}"),
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn test_parse_placeholder_index() {
+ assert_eq!(parse_placeholder_index("$1").unwrap(), 0);
+ assert_eq!(parse_placeholder_index("$2").unwrap(), 1);
+ assert_eq!(parse_placeholder_index("$100").unwrap(), 99);
+ assert!(parse_placeholder_index("$0").is_err());
+ assert!(parse_placeholder_index("$name").is_err());
+ }
+}
diff --git a/datafusion/substrait/src/logical_plan/producer/substrait_producer.rs b/datafusion/substrait/src/logical_plan/producer/substrait_producer.rs
index 51d2c0ca8e..9a6c5317ec 100644
--- a/datafusion/substrait/src/logical_plan/producer/substrait_producer.rs
+++ b/datafusion/substrait/src/logical_plan/producer/substrait_producer.rs
@@ -20,17 +20,17 @@ use crate::logical_plan::producer::{
from_aggregate, from_aggregate_function, from_alias, from_between, from_binary_expr,
from_case, from_cast, from_column, from_distinct, from_empty_relation, from_exists,
from_filter, from_in_list, from_in_subquery, from_join, from_like, from_limit,
- from_literal, from_projection, from_repartition, from_scalar_function,
- from_scalar_subquery, from_set_comparison, from_sort, from_subquery_alias,
- from_table_scan, from_try_cast, from_unary_expr, from_union, from_values,
- from_window, from_window_function, to_substrait_rel, to_substrait_rex,
+ from_literal, from_placeholder, from_projection, from_repartition,
+ from_scalar_function, from_scalar_subquery, from_set_comparison, from_sort,
+ from_subquery_alias, from_table_scan, from_try_cast, from_unary_expr, from_union,
+ from_values, from_window, from_window_function, to_substrait_rel, to_substrait_rex,
};
use datafusion::common::{Column, DFSchemaRef, ScalarValue, substrait_err};
use datafusion::execution::SessionState;
use datafusion::execution::registry::SerializerRegistry;
use datafusion::logical_expr::Subquery;
use datafusion::logical_expr::expr::{
- Alias, Exists, InList, InSubquery, SetComparison, WindowFunction,
+ Alias, Exists, InList, InSubquery, Placeholder, SetComparison, WindowFunction,
};
use datafusion::logical_expr::{
Aggregate, Between, BinaryExpr, Case, Cast, Distinct, EmptyRelation, Expr, Extension,
@@ -388,6 +388,14 @@ pub trait SubstraitProducer: Send + Sync + Sized {
) -> datafusion::common::Result<Expression> {
from_exists(self, exists, schema)
}
+
+ fn handle_placeholder(
+ &mut self,
+ placeholder: &Placeholder,
+ _schema: &DFSchemaRef,
+ ) -> datafusion::common::Result<Expression> {
+ from_placeholder(self, placeholder)
+ }
}
pub struct DefaultSubstraitProducer<'a> {
diff --git a/datafusion/substrait/tests/cases/roundtrip_logical_plan.rs b/datafusion/substrait/tests/cases/roundtrip_logical_plan.rs
index db2f4b587d..1b8496c3dc 100644
--- a/datafusion/substrait/tests/cases/roundtrip_logical_plan.rs
+++ b/datafusion/substrait/tests/cases/roundtrip_logical_plan.rs
@@ -1754,6 +1754,140 @@ async fn roundtrip_read_filter() -> Result<()> {
roundtrip_verify_read_filter_count("SELECT a FROM data where a < 5", 1).await
}
+#[tokio::test]
+async fn roundtrip_placeholder_sql_filter() -> Result<()> {
+ let plan = generate_plan_from_sql("SELECT a, b FROM data WHERE a > $1", false, false)
+ .await?;
+
+ assert_snapshot!(
+ plan,
+ @r"
+ Projection: data.a, data.b
+ Filter: data.a > $1
+ TableScan: data
+ "
+ );
+ Ok(())
+}
+
+#[tokio::test]
+async fn roundtrip_placeholder_sql_projection() -> Result<()> {
+ let plan =
+ generate_plan_from_sql("SELECT a, $1 FROM data WHERE a > $2", false, false)
+ .await?;
+
+ assert_snapshot!(
+ plan,
+ @r"
+ Projection: data.a, $1
+ Filter: data.a > $2
+ TableScan: data
+ "
+ );
+ Ok(())
+}
+
+#[tokio::test]
+async fn roundtrip_placeholder_typed_int64() -> Result<()> {
+ let ctx = create_context().await?;
+
+ let placeholder =
+ Expr::Placeholder(datafusion::logical_expr::expr::Placeholder::new_with_field(
+ "$1".into(),
+ Some(Arc::new(Field::new("$1", DataType::Int64, true))),
+ ));
+ let scan_plan = ctx.table("data").await?.into_optimized_plan()?;
+ let plan = LogicalPlanBuilder::from(scan_plan)
+ .filter(col("a").gt(placeholder))?
+ .build()?;
+
+ let proto = to_substrait_plan(&plan, &ctx.state())?;
+
+ // Verify the producer emits a DynamicParameter in the Substrait proto
+ let plan_rel = proto.relations.first().unwrap();
+ let plan_json = format!("{plan_rel:?}");
+ assert!(
+ plan_json.contains("DynamicParameter"),
+ "Substrait proto should contain DynamicParameter, got: {plan_json}"
+ );
+
+ let plan2 = from_substrait_plan(&ctx.state(), &proto).await?;
+
+ assert_snapshot!(
+ plan2,
+ @r"
+ Filter: data.a > $1
+ TableScan: data
+ "
+ );
+
+ assert_eq!(plan.schema(), plan2.schema());
+ Ok(())
+}
+
+#[tokio::test]
+async fn roundtrip_placeholder_multiple_typed() -> Result<()> {
+ let ctx = create_context().await?;
+
+ let p1 =
+ Expr::Placeholder(datafusion::logical_expr::expr::Placeholder::new_with_field(
+ "$1".into(),
+ Some(Arc::new(Field::new("$1", DataType::Int64, true))),
+ ));
+ let p2 =
+ Expr::Placeholder(datafusion::logical_expr::expr::Placeholder::new_with_field(
+ "$2".into(),
+ Some(Arc::new(Field::new("$2", DataType::Decimal128(5, 2), true))),
+ ));
+ let scan_plan = ctx.table("data").await?.into_optimized_plan()?;
+ let plan = LogicalPlanBuilder::from(scan_plan)
+ .filter(col("a").gt(p1).and(col("b").lt(p2)))?
+ .build()?;
+
+ let proto = to_substrait_plan(&plan, &ctx.state())?;
+ let plan2 = from_substrait_plan(&ctx.state(), &proto).await?;
+
+ assert_snapshot!(
+ plan2,
+ @r"
+ Filter: data.a > $1 AND data.b < $2
+ TableScan: data
+ "
+ );
+
+ assert_eq!(plan.schema(), plan2.schema());
+ Ok(())
+}
+
+#[tokio::test]
+async fn roundtrip_placeholder_typed_utf8() -> Result<()> {
+ let ctx = create_context().await?;
+
+ let placeholder =
+ Expr::Placeholder(datafusion::logical_expr::expr::Placeholder::new_with_field(
+ "$1".into(),
+ Some(Arc::new(Field::new("$1", DataType::Utf8, true))),
+ ));
+ let scan_plan = ctx.table("data").await?.into_optimized_plan()?;
+ let plan = LogicalPlanBuilder::from(scan_plan)
+ .filter(col("f").eq(placeholder))?
+ .build()?;
+
+ let proto = to_substrait_plan(&plan, &ctx.state())?;
+ let plan2 = from_substrait_plan(&ctx.state(), &proto).await?;
+
+ assert_snapshot!(
+ plan2,
+ @r"
+ Filter: data.f = $1
+ TableScan: data
+ "
+ );
+
+ assert_eq!(plan.schema(), plan2.schema());
+ Ok(())
+}
+
fn check_post_join_filters(rel: &Rel) -> Result<()> {
// search for target_rel and field value in proto
match &rel.rel_type {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org