fix(isthmus)!: reject RANGE offsets that cannot be retyped to the ordering column

isthmus/src/main/java/io/substrait/isthmus/expression/WindowBoundConverter.java

@@ -4,6 +4,7 @@
 import io.substrait.expression.ExpressionCreator;
 import io.substrait.expression.WindowBound;
 import io.substrait.isthmus.TypeConverter;
+import io.substrait.type.StringTypeVisitor;
 import io.substrait.type.Type;
 import java.math.BigDecimal;
 import java.math.BigInteger;
@@ -16,7 +17,8 @@
  * Utility for converting Calcite {@link RexWindowBound} to Substrait {@link WindowBound}.
  *
  * <p>Supports {@code CURRENT ROW}, {@code UNBOUNDED}, and {@code PRECEDING}/{@code FOLLOWING}
- * bounds with an arbitrary offset expression.
+ * bounds with an arbitrary offset expression. A RANGE bound's integral literal offset must match
+ * the ordering expression's exact type.
  */
 public class WindowBoundConverter {
 
@@ -31,6 +33,8 @@ public class WindowBoundConverter {
    * @return the corresponding Substrait {@link WindowBound}
    * @throws IllegalStateException if the bound is not one of CURRENT ROW, UNBOUNDED, PRECEDING, or
    *     FOLLOWING
+   * @throws UnsupportedOperationException if a RANGE offset's integral literal does not fit the
+   *     ordering expression's exact type
    */
   public static WindowBound toWindowBound(
       RexWindowBound rexWindowBound,
@@ -45,30 +49,58 @@ public static WindowBound toWindowBound(
     }
 
     RexNode node = rexWindowBound.getOffset();
-    Expression offset =
-        normalizeIntegralOffset(
-            node.accept(rexExpressionConverter),
-            isRows,
-            orderingType,
-            rexExpressionConverter.getTypeConverter());
+    Expression converted = node.accept(rexExpressionConverter);
 
     // Per the spec, zero is not a valid offset; it is equivalent to CurrentRow, and producers
-    // should emit CurrentRow rather than a zero offset_expr.
-    if (integralValue(offset).filter(value -> value == 0).isPresent()) {
+    // should emit CurrentRow rather than a zero offset_expr. Checked before retyping: a zero
+    // offset needs no representation in the ordering expression's type.
+    if (integralValue(converted).filter(value -> value == 0).isPresent()) {
       return WindowBound.CURRENT_ROW;
     }
 
-    if (rexWindowBound.isPreceding()) {
+    // The spec carries a bound's direction in the Preceding/Following choice, not in the sign of
+    // the offset: a negative offset is invalid, and the mirror bound with the magnitude is its
+    // equivalent. Calcite only rejects a negative offset for ROWS, so RANGE reaches here.
+    boolean preceding = rexWindowBound.isPreceding();
+    Optional<Long> negative = integralValue(converted).filter(value -> value < 0);
+    if (negative.isPresent()) {
+      preceding = !preceding;
+      converted = negate(converted, negative.get());
+    }
+
+    Expression offset =
+        normalizeIntegralOffset(
+            converted, isRows, orderingType, rexExpressionConverter.getTypeConverter());
+
+    if (preceding) {
       return WindowBound.Preceding.of(offset);
     }
-    if (rexWindowBound.isFollowing()) {
+    if (rexWindowBound.isFollowing() || negative.isPresent()) {
       return WindowBound.Following.of(offset);
     }
 
     throw new IllegalStateException(
         "window bound was none of CURRENT ROW, UNBOUNDED, PRECEDING or FOLLOWING");
   }
 
+  private static Expression negate(Expression offset, long value) {
+    long negated;
+    try {
+      negated = Math.negateExact(value);
+    } catch (ArithmeticException e) {
+      // Long.MIN_VALUE has no positive long representation.
+      throw new UnsupportedOperationException("window offset " + value + " cannot be negated");
+    }
+    return integralLiteralOfType(offset.getType(), negated)
+        .orElseThrow(
+            () ->
+                new UnsupportedOperationException(
+                    "window offset "
+                        + value
+                        + " cannot be negated within its own type "
+                        + offset.getType().accept(new StringTypeVisitor())));
+  }
+
   private static Expression normalizeIntegralOffset(
       Expression offset,
       boolean isRows,
@@ -82,10 +114,19 @@ private static Expression normalizeIntegralOffset(
       // The spec requires a BOUNDS_TYPE_ROWS offset_expr to be int64.
       return ExpressionCreator.i64(false, value.get());
     }
-    // BOUNDS_TYPE_RANGE: keep add(T, D) -> T defined for the ordering expression's type T.
+    // BOUNDS_TYPE_RANGE: an exact type match is isthmus's own policy, not a spec mandate.
     return orderingType
         .map(typeConverter::toSubstrait)
-        .flatMap(type -> integralLiteralOfType(type, value.get()))
+        .map(
+            type ->
+                integralLiteralOfType(type, value.get())
+                    .orElseThrow(
+                        () ->
+                            new UnsupportedOperationException(
+                                "RANGE window offset "
+                                    + value.get()
+                                    + " does not fit the ordering expression's type "
+                                    + type.accept(new StringTypeVisitor()))))
         .orElse(offset);
   }
 

isthmus/src/test/java/io/substrait/isthmus/WindowBoundConverterTest.java

@@ -1,6 +1,8 @@
 package io.substrait.isthmus;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import io.substrait.expression.Expression;
 import io.substrait.expression.ExpressionCreator;
@@ -83,8 +85,8 @@ void rowsIntegralOffsetIsWidenedToI64() {
 
   @Test
   void rangeIntegralOffsetTakesTheOrderingExpressionType() {
-    // Per the spec, a RANGE offset's type D must keep add(T, D) -> T defined for the ordering
-    // expression's type T -- forcing it to int64 would break that for, e.g., an i32 column.
+    // isthmus requires a RANGE offset's type to exactly match the ordering expression's type T --
+    // forcing it to int64 would break that for, e.g., an i32 column.
     RexNode offset = c(5, SqlTypeName.INTEGER);
     RexWindowBound bound = RexWindowBounds.preceding(offset);
     RelDataType orderingType = t(SqlTypeName.INTEGER);
@@ -96,6 +98,114 @@ void rangeIntegralOffsetTakesTheOrderingExpressionType() {
     assertEquals(WindowBound.Preceding.of(ExpressionCreator.i32(false, 5)), converted);
   }
 
+  @Test
+  void rangeOffsetOutOfRangeForOrderingTypeThrows() {
+    // Calcite's SqlWindow#validateFrameBoundary only checks the bound's type family against the
+    // ordering type for RANGE, not its range, so an offset that doesn't fit the ordering column's
+    // narrower type must be rejected here rather than silently kept as the literal's own type.
+    RexNode offset = c(100000, SqlTypeName.INTEGER);
+    RexWindowBound bound = RexWindowBounds.preceding(offset);
+    RelDataType orderingType = t(SqlTypeName.SMALLINT);
+
+    UnsupportedOperationException ex =
+        assertThrows(
+            UnsupportedOperationException.class,
+            () ->
+                WindowBoundConverter.toWindowBound(
+                    bound, false, Optional.of(orderingType), rexExpressionConverter));
+    assertTrue(ex.getMessage().contains("100000"));
+    assertTrue(ex.getMessage().contains("i16"));
+  }
+
+  @Test
+  void rangeOffsetAcceptsTheOrderingTypesUpperBoundButNotBeyondIt() {
+    // The boundary this narrower-type guard actually enforces: the maximum i16 value retypes
+    // cleanly, but one past it throws instead of silently keeping the literal's own (wider) type.
+    RexWindowBound acceptedBound =
+        RexWindowBounds.preceding(c(Short.MAX_VALUE, SqlTypeName.INTEGER));
+    RelDataType orderingType = t(SqlTypeName.SMALLINT);
+
+    WindowBound converted =
+        WindowBoundConverter.toWindowBound(
+            acceptedBound, false, Optional.of(orderingType), rexExpressionConverter);
+    assertEquals(
+        WindowBound.Preceding.of(ExpressionCreator.i16(false, Short.MAX_VALUE)), converted);
+
+    RexWindowBound rejectedBound =
+        RexWindowBounds.preceding(c(Short.MAX_VALUE + 1, SqlTypeName.INTEGER));
+    assertThrows(
+        UnsupportedOperationException.class,
+        () ->
+            WindowBoundConverter.toWindowBound(
+                rejectedBound, false, Optional.of(orderingType), rexExpressionConverter));
+  }
+
+  @Test
+  void rangeOffsetExceedingDecimalPrecisionThrows() {
+    RexNode offset = c(12345, SqlTypeName.INTEGER);
+    RexWindowBound bound = RexWindowBounds.preceding(offset);
+    RelDataType orderingType = t(SqlTypeName.DECIMAL, 5, 2);
+
+    UnsupportedOperationException ex =
+        assertThrows(
+            UnsupportedOperationException.class,
+            () ->
+                WindowBoundConverter.toWindowBound(
+                    bound, false, Optional.of(orderingType), rexExpressionConverter));
+    assertTrue(ex.getMessage().contains("12345"));
+    assertTrue(ex.getMessage().contains("decimal<5,2>"));
+  }
+
+  @Test
+  void rangeOffsetFailingFloatRoundTripThrows() {
+    // 16_777_217 (2^24 + 1) is the first integer a 24-bit float mantissa cannot represent exactly.
+    RexNode offset = c(16777217, SqlTypeName.INTEGER);
+    RexWindowBound bound = RexWindowBounds.preceding(offset);
+    RelDataType orderingType = t(SqlTypeName.REAL);
+
+    UnsupportedOperationException ex =
+        assertThrows(
+            UnsupportedOperationException.class,
+            () ->
+                WindowBoundConverter.toWindowBound(
+                    bound, false, Optional.of(orderingType), rexExpressionConverter));
+    assertTrue(ex.getMessage().contains("16777217"));
+    assertTrue(ex.getMessage().contains("fp32"));
+  }
+
+  @Test
+  void rangeOffsetAgainstUnsupportedOrderingTypeThrows() {
+    // integralLiteralOfType has no case for a temporal ordering column, so no non-zero offset can
+    // ever be retyped to it.
+    RexNode offset = c(5, SqlTypeName.INTEGER);
+    RexWindowBound bound = RexWindowBounds.preceding(offset);
+    RelDataType orderingType = t(SqlTypeName.TIMESTAMP);
+
+    UnsupportedOperationException ex =
+        assertThrows(
+            UnsupportedOperationException.class,
+            () ->
+                WindowBoundConverter.toWindowBound(
+                    bound, false, Optional.of(orderingType), rexExpressionConverter));
+    assertTrue(ex.getMessage().contains("5"));
+    assertTrue(ex.getMessage().contains("precision_timestamp<"));
+  }
+
+  @Test
+  void rangeOffsetAgainstDateOrderingTypeThrows() {
+    // Same "no such case" path as the TIMESTAMP test above, covered separately so that path isn't
+    // pinned by a single ordering type.
+    RexNode offset = c(5, SqlTypeName.INTEGER);
+    RexWindowBound bound = RexWindowBounds.preceding(offset);
+    RelDataType orderingType = t(SqlTypeName.DATE);
+
+    assertThrows(
+        UnsupportedOperationException.class,
+        () ->
+            WindowBoundConverter.toWindowBound(
+                bound, false, Optional.of(orderingType), rexExpressionConverter));
+  }
+
   @Test
   void zeroOffsetBecomesCurrentRow() {
     // Per the spec, zero is not a valid offset and is equivalent to CurrentRow; producers should
@@ -108,4 +218,78 @@ void zeroOffsetBecomesCurrentRow() {
 
     assertEquals(WindowBound.CURRENT_ROW, converted);
   }
+
+  @Test
+  void zeroOffsetBecomesCurrentRowEvenWhenItWouldNotFitTheDecimalOrderingType() {
+    // Regression test: a zero offset must short-circuit to CurrentRow before retyping is
+    // attempted. digitCount(0) is 1, so retyping 0 against DECIMAL(5,5) would otherwise throw
+    // (1 + scale(5) > precision(5)), even though zero always needs no representation at all.
+    RexNode offset = c(0, SqlTypeName.INTEGER);
+    RexWindowBound bound = RexWindowBounds.preceding(offset);
+    RelDataType orderingType = t(SqlTypeName.DECIMAL, 5, 5);
+
+    WindowBound converted =
+        WindowBoundConverter.toWindowBound(
+            bound, false, Optional.of(orderingType), rexExpressionConverter);
+
+    assertEquals(WindowBound.CURRENT_ROW, converted);
+  }
+
+  @Test
+  void zeroOffsetBecomesCurrentRowEvenAgainstAnUnsupportedOrderingType() {
+    // Regression test: integralLiteralOfType has no case for TIMESTAMP, so retyping a zero offset
+    // against it would otherwise throw, even though zero always needs no representation at all.
+    RexNode offset = c(0, SqlTypeName.INTEGER);
+    RexWindowBound bound = RexWindowBounds.preceding(offset);
+    RelDataType orderingType = t(SqlTypeName.TIMESTAMP);
+
+    WindowBound converted =
+        WindowBoundConverter.toWindowBound(
+            bound, false, Optional.of(orderingType), rexExpressionConverter);
+
+    assertEquals(WindowBound.CURRENT_ROW, converted);
+  }
+
+  @Test
+  void negativePrecedingOffsetIsFlippedToFollowingWithItsMagnitude() {
+    // The spec carries a bound's direction in the Preceding/Following choice, not in the sign of
+    // the offset: RANGE BETWEEN -5 PRECEDING is equivalent to FOLLOWING 5.
+    RexNode offset = c(-5, SqlTypeName.INTEGER);
+    RexWindowBound bound = RexWindowBounds.preceding(offset);
+    RelDataType orderingType = t(SqlTypeName.INTEGER);
+
+    WindowBound converted =
+        WindowBoundConverter.toWindowBound(
+            bound, false, Optional.of(orderingType), rexExpressionConverter);
+
+    assertEquals(WindowBound.Following.of(ExpressionCreator.i32(false, 5)), converted);
+  }
+
+  @Test
+  void negativeFollowingOffsetIsFlippedToPrecedingWithItsMagnitude() {
+    RexNode offset = c(-5, SqlTypeName.INTEGER);
+    RexWindowBound bound = RexWindowBounds.following(offset);
+    RelDataType orderingType = t(SqlTypeName.INTEGER);
+
+    WindowBound converted =
+        WindowBoundConverter.toWindowBound(
+            bound, false, Optional.of(orderingType), rexExpressionConverter);
+
+    assertEquals(WindowBound.Preceding.of(ExpressionCreator.i32(false, 5)), converted);
+  }
+
+  @Test
+  void negativeOffsetOverflowingLongIsRejectedRatherThanThrowingArithmeticException() {
+    // Long.MIN_VALUE has no positive long representation; Math.negateExact would throw
+    // ArithmeticException, which toWindowBound does not document.
+    RexNode offset = c(Long.MIN_VALUE, SqlTypeName.BIGINT);
+    RexWindowBound bound = RexWindowBounds.preceding(offset);
+    RelDataType orderingType = t(SqlTypeName.BIGINT);
+
+    assertThrows(
+        UnsupportedOperationException.class,
+        () ->
+            WindowBoundConverter.toWindowBound(
+                bound, false, Optional.of(orderingType), rexExpressionConverter));
+  }
 }