feat(core): enforce Fetch offset/count expressions are integer-typed

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

@@ -1,6 +1,7 @@
 package io.substrait.relation;
 
 import io.substrait.expression.Expression;
+import io.substrait.type.StringTypeVisitor;
 import io.substrait.type.Type;
 import io.substrait.util.VisitationContext;
 import java.util.Optional;
@@ -29,6 +30,33 @@ public abstract class Fetch extends SingleInputRel implements HasExtension {
    */
   public abstract Optional<Expression> getCount();
 
+  /**
+   * Validates that the offset and count expressions are integer-typed. Both must evaluate to a
+   * non-negative integer; {@code i64} is recommended but not required, so any integer width is
+   * accepted (spec v0.99.0).
+   *
+   * @throws IllegalArgumentException if the offset or count expression is not integer-typed
+   */
+  @Value.Check
+  protected void check() {
+    requireIntegerType(getOffset(), "offset");
+    requireIntegerType(getCount(), "count");
+  }
+
+  private static void requireIntegerType(Optional<Expression> expression, String field) {
+    expression.ifPresent(
+        e -> {
+          Type type = e.getType();
+          if (!type.isInteger()) {
+            throw new IllegalArgumentException(
+                "Fetch "
+                    + field
+                    + " expression must have an integer type (i8, i16, i32 or i64), but got: "
+                    + type.accept(new StringTypeVisitor()));
+          }
+        });
+  }
+
   @Override
   protected Type.Struct deriveRecordType() {
     return getInput().getRecordType();

core/src/main/java/io/substrait/type/Type.java

@@ -53,6 +53,16 @@ default boolean equalsIgnoringNullability(Type other) {
     return TypeCreator.asNullable(this).equals(TypeCreator.asNullable(other));
   }
 
+  /**
+   * Returns whether this is one of the fixed-width signed integer types: {@link I8}, {@link I16},
+   * {@link I32} or {@link I64}.
+   *
+   * @return {@code true} if this is an integer type
+   */
+  default boolean isInteger() {
+    return this instanceof I8 || this instanceof I16 || this instanceof I32 || this instanceof I64;
+  }
+
   /** The boolean type. */
   @Value.Immutable
   abstract class Bool implements Type {

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

@@ -0,0 +1,64 @@
+package io.substrait.relation;
+
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import io.substrait.TestBase;
+import java.util.Arrays;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Validation tests for {@link Fetch}, whose offset/count expressions must have an integer type.
+ * Round-trip coverage lives in the {@code io.substrait.type.proto} package.
+ */
+class FetchTest extends TestBase {
+
+  // Reuse the same schema shape as FetchRoundtripTest so the two files stay in sync.
+  final Rel table =
+      sb.namedScan(Arrays.asList("T"), Arrays.asList("a", "b"), Arrays.asList(R.I64, R.STRING));
+
+  /** Every integer width is an acceptable offset/count expression type. */
+  @Test
+  void integerWidthsAccepted() {
+    fetch().offset(sb.i8(1)).count(sb.i8(2)).build();
+    fetch().offset(sb.i16(1)).count(sb.i16(2)).build();
+    fetch().offset(sb.i32(1)).count(sb.i32(2)).build();
+    fetch().offset(sb.i64(1)).count(sb.i64(2)).build();
+  }
+
+  /** A non-integer offset expression is rejected at construction time. */
+  @Test
+  void nonIntegerOffsetRejected() {
+    assertThrows(IllegalArgumentException.class, () -> fetch().offset(sb.fp64(1.0)).build());
+  }
+
+  /** A non-integer count expression is rejected at construction time. */
+  @Test
+  void nonIntegerCountRejected() {
+    assertThrows(IllegalArgumentException.class, () -> fetch().count(sb.fp64(1.0)).build());
+  }
+
+  /** A non-integer type reaches the check via the proto conversion path too. */
+  @Test
+  void nonIntegerOffsetRejectedViaProto() {
+    // Build a FetchRel proto directly with a non-integer offset_expr so the Fetch POJO check is
+    // exercised by ProtoRelConverter rather than by direct construction.
+    io.substrait.proto.Rel inputProto = relProtoConverter.toProto(table);
+    io.substrait.proto.Expression fp64Expr =
+        io.substrait.proto.Expression.newBuilder()
+            .setLiteral(io.substrait.proto.Expression.Literal.newBuilder().setFp64(1.0).build())
+            .build();
+    io.substrait.proto.FetchRel fetchRel =
+        io.substrait.proto.FetchRel.newBuilder()
+            .setCommon(io.substrait.proto.RelCommon.newBuilder().build())
+            .setInput(inputProto)
+            .setOffsetExpr(fp64Expr)
+            .build();
+    io.substrait.proto.Rel protoRel =
+        io.substrait.proto.Rel.newBuilder().setFetch(fetchRel).build();
+    assertThrows(IllegalArgumentException.class, () -> protoRelConverter.from(protoRel));
+  }
+
+  private ImmutableFetch.Builder fetch() {
+    return Fetch.builder().input(table);
+  }
+}

core/src/test/java/io/substrait/type/TypeTest.java

@@ -0,0 +1,52 @@
+package io.substrait.type;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import org.junit.jupiter.api.Test;
+
+/** Tests for the default methods on {@link Type}. */
+class TypeTest {
+  private static final TypeCreator R = TypeCreator.REQUIRED;
+  private static final TypeCreator N = TypeCreator.NULLABLE;
+
+  @Test
+  void integerWidthsAreIntegers() {
+    assertTrue(R.I8.isInteger());
+    assertTrue(R.I16.isInteger());
+    assertTrue(R.I32.isInteger());
+    assertTrue(R.I64.isInteger());
+  }
+
+  @Test
+  void nullabilityIsIrrelevantForIsInteger() {
+    assertTrue(N.I8.isInteger());
+    assertTrue(N.I16.isInteger());
+    assertTrue(N.I32.isInteger());
+    assertTrue(N.I64.isInteger());
+  }
+
+  @Test
+  void nonIntegerPrimitivesAreNotIntegers() {
+    assertFalse(R.BOOLEAN.isInteger());
+    assertFalse(R.FP32.isInteger());
+    assertFalse(R.FP64.isInteger());
+    assertFalse(R.STRING.isInteger());
+    assertFalse(R.BINARY.isInteger());
+    assertFalse(R.DATE.isInteger());
+    assertFalse(R.UUID.isInteger());
+  }
+
+  @Test
+  void compoundAndSpecialTypesAreNotIntegers() {
+    assertFalse(R.decimal(10, 2).isInteger());
+    assertFalse(R.struct(R.I64).isInteger());
+    assertFalse(R.list(R.I64).isInteger());
+    assertFalse(R.map(R.I64, R.I64).isInteger());
+  }
+
+  @Test
+  void userDefinedIsNotAnInteger() {
+    assertFalse(R.userDefined("urn:test", "t").isInteger());
+  }
+}