fix(core)!: require exactly one ordering expression for RANGE window bounds

core/src/main/java/io/substrait/dsl/SubstraitBuilder.java

@@ -1990,7 +1990,9 @@ public Expression.ScalarFunctionInvocation scalarFn(
   }
 
   /**
-   * Creates a window function invocation with specified arguments and window bounds.
+   * Creates a window function invocation with specified arguments and window bounds. Supplies no
+   * ordering expressions, so a RANGE bound with a Preceding or Following side is rejected outright;
+   * use the {@code sort}-carrying overload for that.
    *
    * @param urn the URN of the extension containing the function
    * @param key the function key (name and signature)
@@ -2013,13 +2015,54 @@ public Expression.WindowFunctionInvocation windowFn(
       WindowBound lowerBound,
       WindowBound upperBound,
       Expression... args) {
+    return windowFn(
+        urn,
+        key,
+        outputType,
+        aggregationPhase,
+        invocation,
+        Collections.emptyList(),
+        boundsType,
+        lowerBound,
+        upperBound,
+        args);
+  }
+
+  /**
+   * Creates a window function invocation with specified arguments, window bounds, and ordering.
+   *
+   * @param urn the URN of the extension containing the function
+   * @param key the function key (name and signature)
+   * @param outputType the output type of the function
+   * @param aggregationPhase the aggregation phase
+   * @param invocation the aggregation invocation mode
+   * @param sort the ordering expressions for the window, required by a RANGE bound with a Preceding
+   *     or Following side
+   * @param boundsType the type of window bounds
+   * @param lowerBound the lower bound of the window
+   * @param upperBound the upper bound of the window
+   * @param args the arguments to pass to the function
+   * @return a new {@link Expression.WindowFunctionInvocation}
+   */
+  public Expression.WindowFunctionInvocation windowFn(
+      String urn,
+      String key,
+      Type outputType,
+      Expression.AggregationPhase aggregationPhase,
+      Expression.AggregationInvocation invocation,
+      List<Expression.SortField> sort,
+      Expression.WindowBoundsType boundsType,
+      WindowBound lowerBound,
+      WindowBound upperBound,
+      Expression... args) {
     SimpleExtension.WindowFunctionVariant declaration =
         extensions.getWindowFunction(SimpleExtension.FunctionAnchor.of(urn, key));
     return Expression.WindowFunctionInvocation.builder()
         .declaration(declaration)
         .outputType(outputType)
         .aggregationPhase(aggregationPhase)
         .invocation(invocation)
+        .sort(sort)
         .boundsType(boundsType)
         .lowerBound(lowerBound)
         .upperBound(upperBound)

core/src/main/java/io/substrait/expression/Expression.java

@@ -1629,8 +1629,9 @@ public Type getType() {
     public abstract AggregationInvocation invocation();
 
     /**
-     * Validates that variadic arguments satisfy the parameter consistency requirement, and that
-     * {@code bounds_type} is set whenever a window bound requires it.
+     * Validates that variadic arguments satisfy the parameter consistency requirement, that {@code
+     * bounds_type} is set whenever a window bound requires it, and that a RANGE bound with a
+     * Preceding or Following side has exactly one, non-CLUSTERED ordering expression.
      *
      * <p>When CONSISTENT, all variadic arguments must have the same type (ignoring nullability).
      * When INCONSISTENT, arguments can have different types.
@@ -1639,6 +1640,8 @@ public Type getType() {
     protected void check() {
       VariadicParameterConsistencyValidator.validate(declaration(), arguments());
       WindowBound.checkBoundsType(boundsType(), lowerBound(), upperBound());
+      WindowBound.checkRangeOrdering(
+          boundsType(), lowerBound(), upperBound(), sort(), declaration().key());
     }
 
     /**

core/src/main/java/io/substrait/expression/WindowBound.java

@@ -1,5 +1,6 @@
 package io.substrait.expression;
 
+import java.util.List;
 import java.util.Optional;
 import org.immutables.value.Value;
 
@@ -62,6 +63,50 @@ static void checkBoundsType(
     }
   }
 
+  /**
+   * Validates a RANGE window's ordering against its bounds, per the spec's rule that a RANGE frame
+   * with a {@link Preceding} or {@link Following} bound must have exactly one ordering expression,
+   * which must not use {@code SORT_DIRECTION_CLUSTERED}.
+   *
+   * @param boundsType the window's bounds type
+   * @param lowerBound the window's lower bound
+   * @param upperBound the window's upper bound
+   * @param sorts the window's ordering expressions
+   * @param function identifies the window function being validated, for the exception message
+   * @throws IllegalArgumentException if {@code boundsType} is {@code RANGE} and either bound is
+   *     {@link Preceding} or {@link Following}, and {@code sorts} does not hold exactly one
+   *     ordering expression whose direction is not {@code SORT_DIRECTION_CLUSTERED}
+   */
+  static void checkRangeOrdering(
+      Expression.WindowBoundsType boundsType,
+      WindowBound lowerBound,
+      WindowBound upperBound,
+      List<Expression.SortField> sorts,
+      String function) {
+    boolean needsSingleOrdering =
+        boundsType == Expression.WindowBoundsType.RANGE
+            && (lowerBound instanceof Preceding
+                || lowerBound instanceof Following
+                || upperBound instanceof Preceding
+                || upperBound instanceof Following);
+    if (!needsSingleOrdering) {
+      return;
+    }
+    if (sorts.size() != 1) {
+      throw new IllegalArgumentException(
+          function
+              + ": a RANGE bound with a Preceding or Following side requires exactly one ordering"
+              + " expression, but found "
+              + sorts.size());
+    }
+    if (sorts.get(0).direction() == Expression.SortDirection.CLUSTERED) {
+      throw new IllegalArgumentException(
+          function
+              + ": a RANGE bound with a Preceding or Following side cannot use"
+              + " SORT_DIRECTION_CLUSTERED for its ordering expression");
+    }
+  }
+
   /**
    * Visitor over the concrete {@link WindowBound} kinds.
    *
@@ -123,9 +168,8 @@ abstract class Preceding implements WindowBound {
     public abstract Expression offset();
 
     /**
-     * Creates a {@link Preceding} bound from a literal row offset. For {@code BOUNDS_TYPE_ROWS}
-     * only: a RANGE bound's offset must be type-compatible with the ordering expression, so use
-     * {@link #of(Expression)} there.
+     * Creates a {@link Preceding} bound from a literal {@code i64} row offset. Valid for ROWS, or
+     * for RANGE over an {@code i64} ordering expression; use {@link #of(Expression)} otherwise.
      *
      * @param offset the row offset preceding the current row
      * @return the preceding bound
@@ -161,9 +205,8 @@ abstract class Following implements WindowBound {
     public abstract Expression offset();
 
     /**
-     * Creates a {@link Following} bound from a literal row offset. For {@code BOUNDS_TYPE_ROWS}
-     * only: a RANGE bound's offset must be type-compatible with the ordering expression, so use
-     * {@link #of(Expression)} there.
+     * Creates a {@link Following} bound from a literal {@code i64} row offset. Valid for ROWS, or
+     * for RANGE over an {@code i64} ordering expression; use {@link #of(Expression)} otherwise.
      *
      * @param offset the row offset following the current row
      * @return the following bound

core/src/main/java/io/substrait/relation/ConsistentPartitionWindow.java

@@ -43,6 +43,24 @@ public abstract class ConsistentPartitionWindow extends SingleInputRel implement
    */
   public abstract List<SortField> getSorts();
 
+  /**
+   * Validates that a RANGE bound with a Preceding or Following side has exactly one, non-CLUSTERED
+   * ordering expression, for every window function invocation.
+   */
+  @Value.Check
+  protected void check() {
+    List<WindowRelFunctionInvocation> windowFunctions = getWindowFunctions();
+    for (int i = 0; i < windowFunctions.size(); i++) {
+      WindowRelFunctionInvocation windowFunction = windowFunctions.get(i);
+      WindowBound.checkRangeOrdering(
+          windowFunction.boundsType(),
+          windowFunction.lowerBound(),
+          windowFunction.upperBound(),
+          getSorts(),
+          "window function " + i + " (" + windowFunction.declaration().key() + ")");
+    }
+  }
+
   /**
    * Derives the output record type by appending window outputs to the input type.
    *

core/src/test/java/io/substrait/dsl/SubstraitBuilderTest.java

@@ -3,11 +3,13 @@
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertInstanceOf;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 
 import io.substrait.TestBase;
 import io.substrait.expression.AggregateFunctionInvocation;
 import io.substrait.expression.Expression;
 import io.substrait.expression.FieldReference;
+import io.substrait.expression.WindowBound;
 import io.substrait.extension.DefaultExtensionCatalog;
 import io.substrait.extension.SimpleExtension;
 import io.substrait.plan.Plan;
@@ -227,6 +229,43 @@ void testBooleanLogic() {
       assertNotNull(builder.not(b1));
       assertNotNull(builder.isNull(b1));
     }
+
+    @Test
+    void testWindowFunctionWithOrdering() {
+      // The single shape the sorts-carrying overload exists for: a RANGE bound with a Preceding
+      // side, which requires exactly one ordering expression.
+      final NamedScan scan = createSimpleScan();
+      final Expression.WindowFunctionInvocation windowFn =
+          builder.windowFn(
+              DefaultExtensionCatalog.FUNCTIONS_ARITHMETIC,
+              "lead:any",
+              Type.I32.builder().nullable(false).build(),
+              Expression.AggregationPhase.INITIAL_TO_RESULT,
+              Expression.AggregationInvocation.ALL,
+              builder.sortFields(scan, 0),
+              Expression.WindowBoundsType.RANGE,
+              WindowBound.Preceding.of(builder.i32(5)),
+              WindowBound.CURRENT_ROW,
+              builder.fieldReference(scan, 0));
+
+      assertNotNull(windowFn);
+      assertEquals(1, windowFn.sort().size());
+
+      // The no-sorts overload cannot express this shape at all.
+      assertThrows(
+          IllegalArgumentException.class,
+          () ->
+              builder.windowFn(
+                  DefaultExtensionCatalog.FUNCTIONS_ARITHMETIC,
+                  "lead:any",
+                  Type.I32.builder().nullable(false).build(),
+                  Expression.AggregationPhase.INITIAL_TO_RESULT,
+                  Expression.AggregationInvocation.ALL,
+                  Expression.WindowBoundsType.RANGE,
+                  WindowBound.Preceding.of(builder.i32(5)),
+                  WindowBound.CURRENT_ROW,
+                  builder.fieldReference(scan, 0)));
+    }
   }
 
   @Nested

core/src/test/java/io/substrait/relation/ExpressionCopyOnWriteVisitorTest.java

@@ -98,7 +98,12 @@ void windowFunctionBoundOffsetsAreRewritten() {
             .declaration(declaration)
             .arguments(Collections.emptyList())
             .partitionBy(Collections.emptyList())
-            .sort(Collections.emptyList())
+            .sort(
+                Collections.singletonList(
+                    Expression.SortField.builder()
+                        .expr(sb.i32(1))
+                        .direction(Expression.SortDirection.ASC_NULLS_FIRST)
+                        .build()))
             .outputType(R.I64)
             .aggregationPhase(Expression.AggregationPhase.INITIAL_TO_RESULT)
             .invocation(Expression.AggregationInvocation.ALL)
@@ -114,6 +119,12 @@ void windowFunctionBoundOffsetsAreRewritten() {
         Optional.of(
             Expression.WindowFunctionInvocation.builder()
                 .from(wfi)
+                .sort(
+                    Collections.singletonList(
+                        Expression.SortField.builder()
+                            .expr(sb.i32(-1))
+                            .direction(Expression.SortDirection.ASC_NULLS_FIRST)
+                            .build()))
                 .lowerBound(WindowBound.Preceding.of(sb.i32(-5)))
                 .upperBound(WindowBound.Following.of(sb.i32(-7)))
                 .build()),

core/src/test/java/io/substrait/relation/OuterReferenceConverterTest.java

@@ -308,7 +308,13 @@ void outerReferenceInsideWindowBoundOffsetIsConverted() {
                                         .declaration(declaration)
                                         .arguments(List.of(sb.fieldReference(input2, 0)))
                                         .partitionBy(Collections.emptyList())
-                                        .sort(Collections.emptyList())
+                                        .sort(
+                                            List.of(
+                                                Expression.SortField.builder()
+                                                    .expr(sb.fieldReference(input2, 0))
+                                                    .direction(
+                                                        Expression.SortDirection.ASC_NULLS_FIRST)
+                                                    .build()))
                                         .outputType(TypeCreator.NULLABLE.I64)
                                         .aggregationPhase(
                                             Expression.AggregationPhase.INITIAL_TO_RESULT)

core/src/test/java/io/substrait/relation/RelCopyOnWriteVisitorTest.java

@@ -61,12 +61,13 @@ private ConsistentPartitionWindow windowOver(Rel input, WindowBound lower, Windo
                     .upperBound(upper)
                     .boundsType(Expression.WindowBoundsType.RANGE)
                     .build()))
+        .sorts(sb.sortFields(input, 0))
         .build();
   }
 
   @Test
   void consistentPartitionWindowBoundOffsetsAreRewritten() {
-    Rel input = sb.namedScan(Arrays.asList("test"), Arrays.asList("a"), Arrays.asList(R.I64));
+    Rel input = sb.namedScan(Arrays.asList("test"), Arrays.asList("a"), Arrays.asList(R.I32));
     ConsistentPartitionWindow window =
         windowOver(input, WindowBound.Preceding.of(sb.i32(5)), WindowBound.Following.of(sb.i32(7)));