chore(deps): bump substrait from 0.63.0 to 0.65.0

datafusion/substrait/Cargo.toml

@@ -42,9 +42,9 @@ half = { workspace = true }
 itertools = { workspace = true }
 object_store = { workspace = true }
 # We need to match the version in substrait, so we don't use the workspace version here
-pbjson-types = { version = "0.8.0" }
+pbjson-types = { version = "0.9.0" }
 prost = { workspace = true }
-substrait = { version = "0.63.0", features = ["serde"] }
+substrait = { version = "0.65.0", features = ["serde"] }
 url = { workspace = true }
 tokio = { workspace = true, features = ["fs"] }
 

datafusion/substrait/src/logical_plan/consumer/expr/field_reference.rs

@@ -74,7 +74,19 @@ fn resolve_outer_reference(
     outer_ref: &substrait::proto::expression::field_reference::OuterReference,
     field_idx: usize,
 ) -> datafusion::common::Result<Expr> {
-    let steps_out = outer_ref.steps_out as usize;
+    use substrait::proto::expression::field_reference::outer_reference::OuterReferenceType;
+    // `StepsOut` is deprecated in favour of `RelReference`, but a relation
+    // reference needs anchors that DataFusion does not assign.
+    #[expect(deprecated)]
+    let steps_out = match outer_ref.outer_reference_type {
+        Some(OuterReferenceType::StepsOut(steps_out)) => steps_out as usize,
+        Some(OuterReferenceType::RelReference(_)) => {
+            return not_impl_err!(
+                "OuterReference by relation reference is not supported"
+            );
+        }
+        None => return substrait_err!("OuterReference without a reference type"),
+    };
     let Some(outer_schema) = consumer.get_outer_schema(steps_out) else {
         return substrait_err!(
             "OuterReference with steps_out={steps_out} \

datafusion/substrait/src/logical_plan/consumer/expr/literal.rs

@@ -25,9 +25,7 @@ use crate::variation_const::{
     INTERVAL_DAY_TIME_TYPE_REF, INTERVAL_MONTH_DAY_NANO_TYPE_NAME,
     INTERVAL_MONTH_DAY_NANO_TYPE_REF, INTERVAL_YEAR_MONTH_TYPE_REF,
     LARGE_CONTAINER_TYPE_VARIATION_REF, TIME_32_TYPE_VARIATION_REF,
-    TIME_64_TYPE_VARIATION_REF, TIMESTAMP_MICRO_TYPE_VARIATION_REF,
-    TIMESTAMP_MILLI_TYPE_VARIATION_REF, TIMESTAMP_NANO_TYPE_VARIATION_REF,
-    TIMESTAMP_SECOND_TYPE_VARIATION_REF, UNSIGNED_INTEGER_TYPE_VARIATION_REF,
+    TIME_64_TYPE_VARIATION_REF, UNSIGNED_INTEGER_TYPE_VARIATION_REF,
     VIEW_CONTAINER_TYPE_VARIATION_REF,
 };
 use datafusion::arrow::array::{AsArray, MapArray, new_empty_array};
@@ -46,7 +44,6 @@ use substrait::proto::expression::Literal;
 use substrait::proto::expression::literal::user_defined::{TypeAnchorType, Val};
 use substrait::proto::expression::literal::{
     IntervalCompound, IntervalDayToSecond, IntervalYearToMonth, LiteralType,
-    interval_day_to_second,
 };
 
 pub async fn from_literal(
@@ -102,28 +99,6 @@ pub(crate) fn from_substrait_literal(
         },
         Some(LiteralType::Fp32(f)) => ScalarValue::Float32(Some(*f)),
         Some(LiteralType::Fp64(f)) => ScalarValue::Float64(Some(*f)),
-        #[expect(deprecated)]
-        Some(LiteralType::Timestamp(t)) => {
-            // Kept for backwards compatibility, new plans should use PrecisionTimestamp(Tz) instead
-            #[expect(deprecated)]
-            match lit.type_variation_reference {
-                TIMESTAMP_SECOND_TYPE_VARIATION_REF => {
-                    ScalarValue::TimestampSecond(Some(*t), None)
-                }
-                TIMESTAMP_MILLI_TYPE_VARIATION_REF => {
-                    ScalarValue::TimestampMillisecond(Some(*t), None)
-                }
-                TIMESTAMP_MICRO_TYPE_VARIATION_REF => {
-                    ScalarValue::TimestampMicrosecond(Some(*t), None)
-                }
-                TIMESTAMP_NANO_TYPE_VARIATION_REF => {
-                    ScalarValue::TimestampNanosecond(Some(*t), None)
-                }
-                others => {
-                    return substrait_err!("Unknown type variation reference {others}");
-                }
-            }
-        }
         Some(LiteralType::PrecisionTimestamp(pt)) => match pt.precision {
             0 => ScalarValue::TimestampSecond(Some(pt.value), None),
             3 => ScalarValue::TimestampMillisecond(Some(pt.value), None),
@@ -381,29 +356,17 @@ pub(crate) fn from_substrait_literal(
             days,
             seconds,
             subseconds,
-            precision_mode,
+            precision,
         })) => {
-            use interval_day_to_second::PrecisionMode;
             // DF only supports millisecond precision, so for any more granular type we lose precision
-            let milliseconds = match precision_mode {
-                #[expect(deprecated)]
-                Some(PrecisionMode::Microseconds(ms)) => ms / 1000,
-                None => {
-                    if *subseconds != 0 {
-                        return substrait_err!(
-                            "Cannot set subseconds field of IntervalDayToSecond without setting precision"
-                        );
-                    } else {
-                        0_i32
-                    }
-                }
-                Some(PrecisionMode::Precision(0)) => *subseconds as i32 * 1000,
-                Some(PrecisionMode::Precision(3)) => *subseconds as i32,
-                Some(PrecisionMode::Precision(6)) => (subseconds / 1000) as i32,
-                Some(PrecisionMode::Precision(9)) => (subseconds / 1000 / 1000) as i32,
-                _ => {
+            let milliseconds = match precision {
+                0 => *subseconds as i32 * 1000,
+                3 => *subseconds as i32,
+                6 => (subseconds / 1000) as i32,
+                9 => (subseconds / 1000 / 1000) as i32,
+                p => {
                     return not_impl_err!(
-                        "Unsupported Substrait interval day to second precision mode: {precision_mode:?}"
+                        "Unsupported Substrait interval day to second precision: {p}"
                     );
                 }
             };
@@ -423,8 +386,7 @@ pub(crate) fn from_substrait_literal(
                     days,
                     seconds,
                     subseconds,
-                    precision_mode:
-                        Some(interval_day_to_second::PrecisionMode::Precision(p)),
+                    precision: p,
                 }),
             ) => {
                 if *p < 0 || *p > 9 {
@@ -607,9 +569,7 @@ mod tests {
                     days: 3,
                     seconds: 4,
                     subseconds: 5,
-                    precision_mode: Some(
-                        interval_day_to_second::PrecisionMode::Precision(6),
-                    ),
+                    precision: 6,
                 }),
             })),
         };

datafusion/substrait/src/logical_plan/consumer/expr/mod.rs

@@ -62,6 +62,9 @@ pub async fn from_substrait_rex(
 ) -> datafusion::common::Result<Expr> {
     match &expression.rex_type {
         Some(t) => match t {
+            RexType::ExecutionContextVariable(_) => {
+                not_impl_err!("Execution context variables are not supported")
+            }
             RexType::Literal(expr) => consumer.consume_literal(expr).await,
             RexType::Selection(expr) => {
                 consumer.consume_field_reference(expr, input_schema).await
@@ -92,8 +95,6 @@ pub async fn from_substrait_rex(
                 consumer.consume_subquery(expr.as_ref(), input_schema).await
             }
             RexType::Nested(expr) => consumer.consume_nested(expr, input_schema).await,
-            #[expect(deprecated)]
-            RexType::Enum(expr) => consumer.consume_enum(expr, input_schema).await,
             RexType::DynamicParameter(expr) => {
                 consumer.consume_dynamic_parameter(expr, input_schema).await
             }
@@ -216,14 +217,14 @@ mod tests {
     async fn window_function_with_range_unit_and_no_order_by()
     -> datafusion::common::Result<()> {
         let substrait = Expression {
-            rex_type: Some(RexType::WindowFunction(
+            rex_type: Some(RexType::WindowFunction(Box::new(
                 substrait::proto::expression::WindowFunction {
                     function_reference: 0,
                     bounds_type: BoundsType::Range as i32,
                     sorts: vec![],
                     ..Default::default()
                 },
-            )),
+            ))),
         };
 
         let mut consumer = test_consumer();
@@ -247,12 +248,12 @@ mod tests {
     #[tokio::test]
     async fn window_function_with_count() -> datafusion::common::Result<()> {
         let substrait = Expression {
-            rex_type: Some(RexType::WindowFunction(
+            rex_type: Some(RexType::WindowFunction(Box::new(
                 substrait::proto::expression::WindowFunction {
                     function_reference: 0,
                     ..Default::default()
                 },
-            )),
+            ))),
         };
 
         let mut consumer = test_consumer();
@@ -274,13 +275,13 @@ mod tests {
     #[tokio::test]
     async fn window_function_with_invalid_invocation() {
         let substrait = Expression {
-            rex_type: Some(RexType::WindowFunction(
+            rex_type: Some(RexType::WindowFunction(Box::new(
                 substrait::proto::expression::WindowFunction {
                     function_reference: 0,
                     invocation: 3,
                     ..Default::default()
                 },
-            )),
+            ))),
         };
 
         let mut consumer = test_consumer();

datafusion/substrait/src/logical_plan/consumer/expr/window_function.rs

@@ -27,11 +27,14 @@ use datafusion::logical_expr::expr::WindowFunctionParams;
 use datafusion::logical_expr::{
     Expr, WindowFrameBound, WindowFrameUnits, WindowFunctionDefinition, expr,
 };
+use substrait::proto::Expression;
 use substrait::proto::aggregate_function::AggregationInvocation;
 use substrait::proto::expression::WindowFunction;
+use substrait::proto::expression::literal::LiteralType;
 use substrait::proto::expression::window_function::{Bound, BoundsType};
 use substrait::proto::expression::{
-    window_function::bound as SubstraitBound, window_function::bound::Kind as BoundKind,
+    Literal, RexType, window_function::bound as SubstraitBound,
+    window_function::bound::Kind as BoundKind,
 };
 
 pub async fn from_window_function(
@@ -85,8 +88,8 @@ pub async fn from_window_function(
     };
     let window_frame = datafusion::logical_expr::WindowFrame::new_bounds(
         bound_units,
-        from_substrait_bound(window.lower_bound.as_ref(), true)?,
-        from_substrait_bound(window.upper_bound.as_ref(), false)?,
+        from_substrait_bound(window.lower_bound.as_deref(), true)?,
+        from_substrait_bound(window.upper_bound.as_deref(), false)?,
     );
 
     window_frame.regularize_order_bys(&mut order_by)?;
@@ -139,20 +142,32 @@ fn from_substrait_bound(
                 BoundKind::CurrentRow(SubstraitBound::CurrentRow {}) => {
                     Ok(WindowFrameBound::CurrentRow)
                 }
-                BoundKind::Preceding(SubstraitBound::Preceding { offset }) => {
-                    if *offset <= 0 {
+                BoundKind::Preceding(bound) => {
+                    #[expect(deprecated)]
+                    let offset =
+                        bound_offset(bound.offset, bound.offset_expr.as_deref())?;
+                    let Some(offset) = offset else {
+                        return Ok(WindowFrameBound::CurrentRow);
+                    };
+                    if offset <= 0 {
                         return plan_err!("Preceding bound must be positive");
                     }
                     Ok(WindowFrameBound::Preceding(ScalarValue::UInt64(Some(
-                        *offset as u64,
+                        offset as u64,
                     ))))
                 }
-                BoundKind::Following(SubstraitBound::Following { offset }) => {
-                    if *offset <= 0 {
+                BoundKind::Following(bound) => {
+                    #[expect(deprecated)]
+                    let offset =
+                        bound_offset(bound.offset, bound.offset_expr.as_deref())?;
+                    let Some(offset) = offset else {
+                        return Ok(WindowFrameBound::CurrentRow);
+                    };
+                    if offset <= 0 {
                         return plan_err!("Following bound must be positive");
                     }
                     Ok(WindowFrameBound::Following(ScalarValue::UInt64(Some(
-                        *offset as u64,
+                        offset as u64,
                     ))))
                 }
                 BoundKind::Unbounded(SubstraitBound::Unbounded {}) => {
@@ -174,3 +189,76 @@ fn from_substrait_bound(
         }
     }
 }
+
+/// Reads the distance of a window frame bound.
+///
+/// The specification requires a consumer to use `offset_expr` when it is set and
+/// to ignore `offset`, and defines a zero `offset_expr` as equivalent to
+/// CurrentRow, which `None` reports here. DataFusion frame bounds hold a
+/// literal, so an expression that is not an int64 literal cannot be
+/// represented.
+fn bound_offset(
+    offset: i64,
+    offset_expr: Option<&Expression>,
+) -> datafusion::common::Result<Option<i64>> {
+    match offset_expr {
+        Some(Expression {
+            rex_type:
+                Some(RexType::Literal(Literal {
+                    literal_type: Some(LiteralType::I64(value)),
+                    ..
+                })),
+        }) => Ok((*value != 0).then_some(*value)),
+        Some(_) => not_impl_err!(
+            "Window frame bound offsets other than int64 literals are not supported"
+        ),
+        None => Ok(Some(offset)),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn i64_literal(value: i64) -> Expression {
+        Expression {
+            rex_type: Some(RexType::Literal(Literal {
+                literal_type: Some(LiteralType::I64(value)),
+                ..Default::default()
+            })),
+        }
+    }
+
+    /// A zero `offset_expr` is defined as equivalent to CurrentRow, so it is
+    /// read as such rather than rejected as a non-positive distance.
+    #[test]
+    fn zero_offset_expression_reads_as_current_row() {
+        #[expect(deprecated)]
+        let bound = Bound {
+            kind: Some(BoundKind::Preceding(Box::new(SubstraitBound::Preceding {
+                offset: 0,
+                offset_expr: Some(Box::new(i64_literal(0))),
+            }))),
+        };
+        assert_eq!(
+            from_substrait_bound(Some(&bound), true).unwrap(),
+            WindowFrameBound::CurrentRow
+        );
+    }
+
+    /// When `offset_expr` is set the consumer must use it and ignore `offset`.
+    #[test]
+    fn offset_expression_wins_over_the_deprecated_offset() {
+        #[expect(deprecated)]
+        let bound = Bound {
+            kind: Some(BoundKind::Preceding(Box::new(SubstraitBound::Preceding {
+                offset: 7,
+                offset_expr: Some(Box::new(i64_literal(3))),
+            }))),
+        };
+        assert_eq!(
+            from_substrait_bound(Some(&bound), true).unwrap(),
+            WindowFrameBound::Preceding(ScalarValue::UInt64(Some(3)))
+        );
+    }
+}