fix!: ensure predicates are parsable

rust/src/delta_datafusion.rs → rust/src/delta_datafusion/mod.rs

@@ -76,6 +76,8 @@ use crate::{open_table, open_table_with_storage_options, DeltaTable, Invariant,
 
 const PATH_COLUMN: &str = "__delta_rs_path";
 
+pub mod expr;
+
 impl From<DeltaTableError> for DataFusionError {
  fn from(err: DeltaTableError) -> Self {
  match err {

rust/src/operations/delete.rs

@@ -20,6 +20,7 @@
 use std::sync::Arc;
 use std::time::{Instant, SystemTime, UNIX_EPOCH};
 
+use crate::delta_datafusion::expr::fmt_expr_to_sql;
 use crate::protocol::{Action, Add, Remove};
 use datafusion::execution::context::{SessionContext, SessionState};
 use datafusion::physical_expr::create_physical_expr;
@@ -263,7 +264,7 @@ async fn execute(
  // Do not make a commit when there are zero updates to the state
  if !actions.is_empty() {
  let operation = DeltaOperation::Delete {
- predicate: Some(predicate.canonical_name()),
+ predicate: Some(fmt_expr_to_sql(&predicate)?),
  };
  version = commit(
  object_store.as_ref(),
@@ -298,7 +299,9 @@ impl std::future::IntoFuture for DeleteBuilder {
  let predicate = match this.predicate {
  Some(predicate) => match predicate {
  Expression::DataFusion(expr) => Some(expr),
- Expression::String(s) => Some(this.snapshot.parse_predicate_expression(s)?),
+ Expression::String(s) => {
+ Some(this.snapshot.parse_predicate_expression(s, &state)?)
+ }
  },
  None => None,
  };
@@ -335,6 +338,7 @@ mod tests {
  use arrow::record_batch::RecordBatch;
  use datafusion::assert_batches_sorted_eq;
  use datafusion::prelude::*;
+ use serde_json::json;
  use std::sync::Arc;
 
  async fn setup_table(partitions: Option<Vec<&str>>) -> DeltaTable {
@@ -456,7 +460,7 @@ mod tests {
  assert_eq!(table.version(), 2);
  assert_eq!(table.get_file_uris().count(), 2);
 
- let (table, metrics) = DeltaOps(table)
+ let (mut table, metrics) = DeltaOps(table)
  .delete()
  .with_predicate(col("value").eq(lit(1)))
  .await
@@ -470,6 +474,11 @@ mod tests {
  assert_eq!(metrics.num_deleted_rows, Some(1));
  assert_eq!(metrics.num_copied_rows, Some(3));
 
+ let commit_info = table.history(None).await.unwrap();
+ let last_commit = &commit_info[commit_info.len() - 1];
+ let parameters = last_commit.operation_parameters.clone().unwrap();
+ assert_eq!(parameters["predicate"], json!("value = 1"));
+
  let expected = vec![
  "+----+-------+------------+",
  "| id | value | modified |",

rust/src/operations/merge.rs

@@ -62,6 +62,7 @@ use serde_json::{Map, Value};
 
 use super::datafusion_utils::{into_expr, maybe_into_expr, Expression};
 use super::transaction::commit;
+use crate::delta_datafusion::expr::fmt_expr_to_sql;
 use crate::delta_datafusion::{parquet_scan_from_actions, register_store};
 use crate::operations::datafusion_utils::MetricObserverExec;
 use crate::operations::write::write_execution_plan;
@@ -171,6 +172,7 @@ impl MergeBuilder {
  let builder = builder(UpdateBuilder::default());
  let op = MergeOperation::try_new(
  &self.snapshot,
+ &self.state.as_ref(),
  builder.predicate,
  builder.updates,
  OperationType::Update,
@@ -204,6 +206,7 @@ impl MergeBuilder {
  let builder = builder(DeleteBuilder::default());
  let op = MergeOperation::try_new(
  &self.snapshot,
+ &self.state.as_ref(),
  builder.predicate,
  HashMap::default(),
  OperationType::Delete,
@@ -240,6 +243,7 @@ impl MergeBuilder {
  let builder = builder(InsertBuilder::default());
  let op = MergeOperation::try_new(
  &self.snapshot,
+ &self.state.as_ref(),
  builder.predicate,
  builder.set,
  OperationType::Insert,
@@ -278,6 +282,7 @@ impl MergeBuilder {
  let builder = builder(UpdateBuilder::default());
  let op = MergeOperation::try_new(
  &self.snapshot,
+ &self.state.as_ref(),
  builder.predicate,
  builder.updates,
  OperationType::Update,
@@ -311,6 +316,7 @@ impl MergeBuilder {
  let builder = builder(DeleteBuilder::default());
  let op = MergeOperation::try_new(
  &self.snapshot,
+ &self.state.as_ref(),
  builder.predicate,
  HashMap::default(),
  OperationType::Delete,
@@ -448,15 +454,21 @@ struct MergeOperation {
 impl MergeOperation {
  pub fn try_new(
  snapshot: &DeltaTableState,
+ state: &Option<&SessionState>,
  predicate: Option<Expression>,
  operations: HashMap<Column, Expression>,
  r#type: OperationType,
  ) -> DeltaResult<Self> {
- let predicate = maybe_into_expr(predicate, snapshot)?;
+ let context = SessionContext::new();
+ let mut s = &context.state();
+ if let Some(df_state) = state {
+ s = df_state;
+ }
+ let predicate = maybe_into_expr(predicate, snapshot, s)?;
  let mut _operations = HashMap::new();
 
  for (column, expr) in operations {
- _operations.insert(column, into_expr(expr, snapshot)?);
+ _operations.insert(column, into_expr(expr, snapshot, s)?);
  }
 
  Ok(MergeOperation {
@@ -518,7 +530,7 @@ async fn execute(
 
  let predicate = match predicate {
  Expression::DataFusion(expr) => expr,
- Expression::String(s) => snapshot.parse_predicate_expression(s)?,
+ Expression::String(s) => snapshot.parse_predicate_expression(s, &state)?,
  };
 
  let schema = snapshot.input_schema()?;
@@ -675,7 +687,10 @@ async fn execute(
  };
 
  let action_type = action_type.to_string();
- let predicate = op.predicate.map(|expr| expr.display_name().unwrap());
+ let predicate = op
+ .predicate
+ .map(|expr| fmt_expr_to_sql(&expr))
+ .transpose()?;
 
  predicates.push(MergePredicate {
  action_type,
@@ -1035,7 +1050,7 @@ async fn execute(
  // Do not make a commit when there are zero updates to the state
  if !actions.is_empty() {
  let operation = DeltaOperation::Merge {
- predicate: Some(predicate.canonical_name()),
+ predicate: Some(fmt_expr_to_sql(&predicate)?),
  matched_predicates: match_operations,
  not_matched_predicates: not_match_target_operations,
  not_matched_by_source_predicates: not_match_source_operations,
@@ -1222,10 +1237,9 @@ mod tests {
  parameters["notMatchedPredicates"],
  json!(r#"[{"actionType":"insert"}]"#)
  );
- // Todo: Expected this predicate to actually be 'value = 1'. Predicate should contain a valid sql expression
  assert_eq!(
  parameters["notMatchedBySourcePredicates"],
- json!(r#"[{"actionType":"update","predicate":"value = Int32(1)"}]"#)
+ json!(r#"[{"actionType":"update","predicate":"value = 1"}]"#)
  );
 
  let expected = vec![
@@ -1447,7 +1461,7 @@ mod tests {
  assert_eq!(parameters["predicate"], json!("id = source.id"));
  assert_eq!(
  parameters["matchedPredicates"],
- json!(r#"[{"actionType":"delete","predicate":"source.value <= Int32(10)"}]"#)
+ json!(r#"[{"actionType":"delete","predicate":"source.value <= 10"}]"#)
  );
 
  let expected = vec![
@@ -1579,7 +1593,7 @@ mod tests {
  assert_eq!(parameters["predicate"], json!("id = source.id"));
  assert_eq!(
  parameters["notMatchedBySourcePredicates"],
- json!(r#"[{"actionType":"delete","predicate":"modified > Utf8(\"2021-02-01\")"}]"#)
+ json!(r#"[{"actionType":"delete","predicate":"modified > '2021-02-01'"}]"#)
  );
 
  let expected = vec![

rust/src/operations/mod.rs

@@ -205,6 +205,7 @@ mod datafusion_utils {
  use arrow_schema::SchemaRef;
  use datafusion::arrow::record_batch::RecordBatch;
  use datafusion::error::Result as DataFusionResult;
+ use datafusion::execution::context::SessionState;
  use datafusion::physical_plan::DisplayAs;
  use datafusion::physical_plan::{
  metrics::{ExecutionPlanMetricsSet, MetricsSet},
@@ -240,19 +241,24 @@ mod datafusion_utils {
  }
  }
 
- pub(crate) fn into_expr(expr: Expression, snapshot: &DeltaTableState) -> DeltaResult<Expr> {
+ pub(crate) fn into_expr(
+ expr: Expression,
+ snapshot: &DeltaTableState,
+ df_state: &SessionState,
+ ) -> DeltaResult<Expr> {
  match expr {
  Expression::DataFusion(expr) => Ok(expr),
- Expression::String(s) => snapshot.parse_predicate_expression(s),
+ Expression::String(s) => snapshot.parse_predicate_expression(s, df_state),
  }
  }
 
  pub(crate) fn maybe_into_expr(
  expr: Option<Expression>,
  snapshot: &DeltaTableState,
+ df_state: &SessionState,
  ) -> DeltaResult<Option<Expr>> {
  Ok(match expr {
- Some(predicate) => Some(into_expr(predicate, snapshot)?),
+ Some(predicate) => Some(into_expr(predicate, snapshot, df_state)?),
  None => None,
  })
  }

rust/src/operations/transaction/conflict_checker.rs

@@ -114,8 +114,11 @@ impl<'a> TransactionInfo<'a> {
  actions: &'a Vec<Action>,
  read_whole_table: bool,
  ) -> DeltaResult<Self> {
+ use datafusion::prelude::SessionContext;
+
+ let session = SessionContext::new();
  let read_predicates = read_predicates
- .map(|pred| read_snapshot.parse_predicate_expression(pred))
+ .map(|pred| read_snapshot.parse_predicate_expression(pred, &session.state()))
  .transpose()?;
  Ok(Self {
  txn_id: "".into(),

rust/src/operations/transaction/state.rs

@@ -5,6 +5,7 @@ use arrow::datatypes::{
  DataType, Field as ArrowField, Schema as ArrowSchema, SchemaRef as ArrowSchemaRef,
 };
 use datafusion::datasource::physical_plan::wrap_partition_type_in_dict;
+use datafusion::execution::context::SessionState;
 use datafusion::optimizer::utils::conjunction;
 use datafusion::physical_optimizer::pruning::{PruningPredicate, PruningStatistics};
 use datafusion_common::config::ConfigOptions;
@@ -104,7 +105,11 @@ impl DeltaTableState {
  }
 
  /// Parse an expression string into a datafusion [`Expr`]
- pub fn parse_predicate_expression(&self, expr: impl AsRef<str>) -> DeltaResult<Expr> {
+ pub fn parse_predicate_expression(
+ &self,
+ expr: impl AsRef<str>,
+ df_state: &SessionState,
+ ) -> DeltaResult<Expr> {
  let dialect = &GenericDialect {};
  let mut tokenizer = Tokenizer::new(dialect, expr.as_ref());
  let tokens = tokenizer
@@ -121,7 +126,7 @@ impl DeltaTableState {
 
  // TODO should we add the table name as qualifier when available?
  let df_schema = DFSchema::try_from_qualified_schema("", self.arrow_schema()?.as_ref())?;
- let context_provider = DummyContextProvider::default();
+ let context_provider = DeltaContextProvider { state: df_state };
  let sql_to_rel = SqlToRel::new(&context_provider);
 
  Ok(sql_to_rel.sql_to_expr(sql, &df_schema, &mut Default::default())?)
@@ -342,59 +347,63 @@ impl PruningStatistics for DeltaTableState {
  }
 }
 
-#[derive(Default)]
-struct DummyContextProvider {
- options: ConfigOptions,
+pub(crate) struct DeltaContextProvider<'a> {
+ state: &'a SessionState,
 }
 
-impl ContextProvider for DummyContextProvider {
+impl<'a> ContextProvider for DeltaContextProvider<'a> {
  fn get_table_provider(&self, _name: TableReference) -> DFResult<Arc<dyn TableSource>> {
  unimplemented!()
  }
 
- fn get_function_meta(&self, _name: &str) -> Option<Arc<ScalarUDF>> {
- unimplemented!()
+ fn get_function_meta(&self, name: &str) -> Option<Arc<ScalarUDF>> {
+ self.state.scalar_functions().get(name).cloned()
  }
 
- fn get_aggregate_meta(&self, _name: &str) -> Option<Arc<AggregateUDF>> {
- unimplemented!()
+ fn get_aggregate_meta(&self, name: &str) -> Option<Arc<AggregateUDF>> {
+ self.state.aggregate_functions().get(name).cloned()
  }
 
- fn get_variable_type(&self, _: &[String]) -> Option<DataType> {
+ fn get_variable_type(&self, _var: &[String]) -> Option<DataType> {
  unimplemented!()
  }
 
  fn options(&self) -> &ConfigOptions {
- &self.options
+ self.state.config_options()
  }
 
- fn get_window_meta(&self, _name: &str) -> Option<Arc<datafusion_expr::WindowUDF>> {
- unimplemented!()
+ fn get_window_meta(&self, name: &str) -> Option<Arc<datafusion_expr::WindowUDF>> {
+ self.state.window_functions().get(name).cloned()
  }
 }
 
 #[cfg(test)]
 mod tests {
  use super::*;
  use crate::operations::transaction::test_utils::{create_add_action, init_table_actions};
+ use datafusion::prelude::SessionContext;
  use datafusion_expr::{col, lit};
 
  #[test]
  fn test_parse_predicate_expression() {
- let state = DeltaTableState::from_actions(init_table_actions(), 0).unwrap();
+ let snapshot = DeltaTableState::from_actions(init_table_actions(), 0).unwrap();
+ let session = SessionContext::new();
+ let state = session.state();
 
  // parses simple expression
- let parsed = state.parse_predicate_expression("value > 10").unwrap();
+ let parsed = snapshot
+ .parse_predicate_expression("value > 10", &state)
+ .unwrap();
  let expected = col("value").gt(lit::<i64>(10));
  assert_eq!(parsed, expected);
 
  // fails for unknown column
- let parsed = state.parse_predicate_expression("non_existent > 10");
+ let parsed = snapshot.parse_predicate_expression("non_existent > 10", &state);
  assert!(parsed.is_err());
 
  // parses complex expression
- let parsed = state
- .parse_predicate_expression("value > 10 OR value <= 0")
+ let parsed = snapshot
+ .parse_predicate_expression("value > 10 OR value <= 0", &state)
  .unwrap();
  let expected = col("value")
  .gt(lit::<i64>(10))