-
Notifications
You must be signed in to change notification settings - Fork 1.2k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Support named query parameters #8384
Changes from 27 commits
7afeb8b
6332bec
cc5e0c7
a114310
928c811
839093e
a836cde
5648dc7
a670409
22894a3
73a59d2
46409c2
8a86a4c
cf5c584
62ae9b9
da02fa2
d98eb2e
79e7216
ba51abd
2468f52
180c303
68980ba
9411940
ba28346
df0942f
d6381c2
69a2080
49e617e
5b8c3b7
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,133 @@ | ||
// 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::error::{_internal_err, _plan_err}; | ||
use crate::{DataFusionError, Result, ScalarValue}; | ||
use arrow_schema::DataType; | ||
use std::collections::HashMap; | ||
|
||
/// The parameter value corresponding to the placeholder | ||
#[derive(Debug, Clone)] | ||
pub enum ParamValues { | ||
/// for positional query parameters, like select * from test where a > $1 and b = $2 | ||
LIST(Vec<ScalarValue>), | ||
/// for named query parameters, like select * from test where a > $foo and b = $goo | ||
MAP(HashMap<String, ScalarValue>), | ||
} | ||
|
||
impl ParamValues { | ||
/// Verify parameter list length and type | ||
pub fn verify(&self, expect: &Vec<DataType>) -> Result<()> { | ||
match self { | ||
ParamValues::LIST(list) => { | ||
// Verify if the number of params matches the number of values | ||
if expect.len() != list.len() { | ||
return _plan_err!( | ||
"Expected {} parameters, got {}", | ||
expect.len(), | ||
list.len() | ||
); | ||
} | ||
|
||
// Verify if the types of the params matches the types of the values | ||
let iter = expect.iter().zip(list.iter()); | ||
for (i, (param_type, value)) in iter.enumerate() { | ||
if *param_type != value.data_type() { | ||
return _plan_err!( | ||
"Expected parameter of type {:?}, got {:?} at index {}", | ||
param_type, | ||
value.data_type(), | ||
i | ||
); | ||
} | ||
} | ||
Ok(()) | ||
} | ||
ParamValues::MAP(_) => { | ||
// If it is a named query, variables can be reused, | ||
// but the lengths are not necessarily equal | ||
Ok(()) | ||
} | ||
} | ||
} | ||
|
||
pub fn get_placeholders_with_values( | ||
&self, | ||
id: &String, | ||
data_type: &Option<DataType>, | ||
) -> Result<ScalarValue> { | ||
match self { | ||
ParamValues::LIST(list) => { | ||
if id.is_empty() || id == "$0" { | ||
return _plan_err!("Empty placeholder id"); | ||
} | ||
// convert id (in format $1, $2, ..) to idx (0, 1, ..) | ||
let idx = id[1..].parse::<usize>().map_err(|e| { | ||
DataFusionError::Internal(format!( | ||
"Failed to parse placeholder id: {e}" | ||
)) | ||
})? - 1; | ||
// value at the idx-th position in param_values should be the value for the placeholder | ||
let value = list.get(idx).ok_or_else(|| { | ||
DataFusionError::Internal(format!( | ||
"No value found for placeholder with id {id}" | ||
)) | ||
})?; | ||
// check if the data type of the value matches the data type of the placeholder | ||
if Some(value.data_type()) != *data_type { | ||
return _internal_err!( | ||
"Placeholder value type mismatch: expected {:?}, got {:?}", | ||
data_type, | ||
value.data_type() | ||
); | ||
} | ||
Ok(value.clone()) | ||
} | ||
ParamValues::MAP(map) => { | ||
// convert name (in format $a, $b, ..) to mapped values (a, b, ..) | ||
let name = &id[1..]; | ||
// value at the name position in param_values should be the value for the placeholder | ||
let value = map.get(name).ok_or_else(|| { | ||
DataFusionError::Internal(format!( | ||
"No value found for placeholder with name {id}" | ||
)) | ||
})?; | ||
// check if the data type of the value matches the data type of the placeholder | ||
if Some(value.data_type()) != *data_type { | ||
return _internal_err!( | ||
"Placeholder value type mismatch: expected {:?}, got {:?}", | ||
data_type, | ||
value.data_type() | ||
); | ||
} | ||
Ok(value.clone()) | ||
} | ||
} | ||
} | ||
} | ||
|
||
impl From<Vec<ScalarValue>> for ParamValues { | ||
fn from(value: Vec<ScalarValue>) -> Self { | ||
Self::LIST(value) | ||
} | ||
} | ||
|
||
impl From<HashMap<String, ScalarValue>> for ParamValues { | ||
fn from(value: HashMap<String, ScalarValue>) -> Self { | ||
Self::MAP(value) | ||
} | ||
} |
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -32,11 +32,12 @@ use datafusion_common::file_options::csv_writer::CsvWriterOptions; | |
use datafusion_common::file_options::json_writer::JsonWriterOptions; | ||
use datafusion_common::parsers::CompressionTypeVariant; | ||
use datafusion_common::{ | ||
DataFusionError, FileType, FileTypeWriterOptions, SchemaError, UnnestOptions, | ||
DataFusionError, FileType, FileTypeWriterOptions, ParamValues, SchemaError, | ||
UnnestOptions, | ||
}; | ||
use datafusion_expr::dml::CopyOptions; | ||
|
||
use datafusion_common::{Column, DFSchema, ScalarValue}; | ||
use datafusion_common::{Column, DFSchema}; | ||
use datafusion_expr::{ | ||
avg, count, is_null, max, median, min, stddev, utils::COUNT_STAR_EXPANSION, | ||
TableProviderFilterPushDown, UNNAMED_TABLE, | ||
|
@@ -1214,7 +1215,7 @@ impl DataFrame { | |
/// .with_param_values(vec![ | ||
/// // value at index 0 --> $1 | ||
/// ScalarValue::from(2i64) | ||
/// ])? | ||
/// ].into())? | ||
/// .collect() | ||
/// .await?; | ||
/// assert_batches_eq!( | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Could you please also add a demonstration of how to use the hashmap version too so that it is easier to discover by a casual user? Perhaps add this to the example: /// // Note you can also provide named parameters
/// let results = ctx
/// .sql("SELECT a FROM example WHERE b = $my_param")
/// .await?
/// // replace $my_param with value 2
/// // Note you can also use a HashMap as well
/// .with_param_values(vec![
/// "my_param",
/// ScalarValue::from(2i64)
/// ])?
/// .collect()
/// .await?;
/// assert_batches_eq!( There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Thanks for your comment, I'm done |
||
|
@@ -1230,8 +1231,8 @@ impl DataFrame { | |
/// # Ok(()) | ||
/// # } | ||
/// ``` | ||
pub fn with_param_values(self, param_values: Vec<ScalarValue>) -> Result<Self> { | ||
let plan = self.plan.with_param_values(param_values)?; | ||
pub fn with_param_values(self, query_values: ParamValues) -> Result<Self> { | ||
let plan = self.plan.with_param_values(query_values)?; | ||
Ok(Self::new(self.session_state, plan)) | ||
} | ||
|
||
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I think it is more appropriate to encapsulate it into a struct