Emit CurrentRow for a zero window frame distance, read a zero offset_expr as CurrentRow, and read RelCommon from UpdateRel

Author: alexandrefimov

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

@@ -146,6 +146,9 @@ fn from_substrait_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");
                     }
@@ -157,6 +160,9 @@ fn from_substrait_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");
                     }
@@ -187,23 +193,72 @@ 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`. DataFusion frame bounds hold a literal, so an expression
-/// that is not an int64 literal cannot be represented.
+/// 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<i64> {
+) -> datafusion::common::Result<Option<i64>> {
     match offset_expr {
         Some(Expression {
             rex_type:
                 Some(RexType::Literal(Literal {
                     literal_type: Some(LiteralType::I64(value)),
                     ..
                 })),
-        }) => Ok(*value),
+        }) => Ok((*value != 0).then_some(*value)),
         Some(_) => not_impl_err!(
             "Window frame bound offsets other than int64 literals are not supported"
         ),
-        None => Ok(offset),
+        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)))
+        );
     }
 }

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

@@ -157,7 +157,7 @@ fn retrieve_rel_common(rel: &Rel) -> Option<&RelCommon> {
             RelType::Window(w) => w.common.as_ref(),
             RelType::Exchange(e) => e.common.as_ref(),
             RelType::Expand(e) => e.common.as_ref(),
-            RelType::Update(_) => None,
+            RelType::Update(u) => u.common.as_ref(),
         },
     }
 }

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

@@ -154,28 +154,39 @@ fn to_substrait_bound(bound: &WindowFrameBound) -> datafusion::common::Result<Bo
         }),
         WindowFrameBound::Preceding(s) => {
             let offset = to_substrait_bound_offset(s)?;
+            if offset == 0 {
+                // A zero distance is equivalent to CurrentRow, and `offset`
+                // cannot represent zero, so the specification asks producers to
+                // emit CurrentRow instead of a zero bound.
+                return Ok(Bound {
+                    kind: Some(BoundKind::CurrentRow(SubstraitBound::CurrentRow {})),
+                });
+            }
             #[expect(deprecated)]
             Ok(Bound {
                 kind: Some(BoundKind::Preceding(Box::new(SubstraitBound::Preceding {
-                    // `offset` carries the int64-literal equivalent for consumers
-                    // that do not read `offset_expr` yet. A zero distance is
-                    // equivalent to CurrentRow and `offset` cannot represent it,
-                    // so the specification asks producers not to write a zero
-                    // `offset_expr`.
+                    // `offset` carries the int64-literal equivalent for
+                    // consumers that do not read `offset_expr` yet.
                     offset,
-                    offset_expr: (offset != 0)
-                        .then(|| Box::new(bound_offset_expr(offset))),
+                    offset_expr: Some(Box::new(bound_offset_expr(offset))),
                 }))),
             })
         }
         WindowFrameBound::Following(s) => {
             let offset = to_substrait_bound_offset(s)?;
+            if offset == 0 {
+                // A zero distance is equivalent to CurrentRow, and `offset`
+                // cannot represent zero, so the specification asks producers to
+                // emit CurrentRow instead of a zero bound.
+                return Ok(Bound {
+                    kind: Some(BoundKind::CurrentRow(SubstraitBound::CurrentRow {})),
+                });
+            }
             #[expect(deprecated)]
             Ok(Bound {
                 kind: Some(BoundKind::Following(Box::new(SubstraitBound::Following {
                     offset,
-                    offset_expr: (offset != 0)
-                        .then(|| Box::new(bound_offset_expr(offset))),
+                    offset_expr: Some(Box::new(bound_offset_expr(offset))),
                 }))),
             })
         }
@@ -216,6 +227,24 @@ mod tests {
     use super::*;
     use datafusion::common::assert_contains;
 
+    #[test]
+    fn zero_distance_bounds_become_current_row() {
+        // `offset` cannot represent zero and the specification asks producers to
+        // emit CurrentRow rather than a zero bound, so neither field is written.
+        let frame = WindowFrame::new_bounds(
+            WindowFrameUnits::Rows,
+            WindowFrameBound::Preceding(ScalarValue::UInt64(Some(0))),
+            WindowFrameBound::Following(ScalarValue::UInt64(Some(0))),
+        );
+        let (lower, upper) = to_substrait_bounds(&frame).unwrap();
+        for bound in [lower, upper] {
+            assert_eq!(
+                bound.kind,
+                Some(BoundKind::CurrentRow(SubstraitBound::CurrentRow {}))
+            );
+        }
+    }
+
     #[test]
     #[expect(deprecated)]
     fn window_frame_offsets() {