fix: handle LIMIT ALL queries using a sentinel value in FetchRel::count

FetchRels with no count set could not be distinguished from FetchRels
with count set to 0.

To account for this, the spec now uses -1 as a sentinel value for this.

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

@@ -353,7 +353,12 @@ private VirtualTableScan newVirtualTable(ReadRel rel) {
 
   private Fetch newFetch(FetchRel rel) {
     var input = from(rel.getInput());
-    var builder = Fetch.builder().input(input).count(rel.getCount()).offset(rel.getOffset());
+    var builder = Fetch.builder().input(input).offset(rel.getOffset());
+    if (rel.getCount() != -1) {
+      // -1 is used as a sentinel value to signal LIMIT ALL
+      // count only needs to be set when it is not -1
+      builder.count(rel.getCount());
+    }
 
     builder
         .commonExtension(optionalAdvancedExtension(rel.getCommon()))

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

@@ -149,9 +149,9 @@ public Rel visit(Fetch fetch) throws RuntimeException {
         FetchRel.newBuilder()
             .setCommon(common(fetch))
             .setInput(toProto(fetch.getInput()))
-            .setOffset(fetch.getOffset());
-
-    fetch.getCount().ifPresent(f -> builder.setCount(f));
+            .setOffset(fetch.getOffset())
+            // -1 is used as a sentinel value to signal LIMIT ALL
+            .setCount(fetch.getCount().orElse(-1));
 
     fetch.getExtension().ifPresent(ae -> builder.setAdvancedExtension(ae.toProto()));
     return Rel.newBuilder().setFetch(builder).build();

isthmus/src/main/java/io/substrait/isthmus/SubstraitRelVisitor.java

@@ -29,6 +29,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.OptionalLong;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import org.apache.calcite.rel.RelFieldCollation;
@@ -283,22 +284,30 @@ public Rel visit(org.apache.calcite.rel.core.Match match) {
 
   @Override
   public Rel visit(org.apache.calcite.rel.core.Sort sort) {
-    var input = apply(sort.getInput());
-    var fields =
-        sort.getCollation().getFieldCollations().stream()
-            .map(t -> toSortField(t, input.getRecordType()))
-            .collect(java.util.stream.Collectors.toList());
-    var convertedSort = Sort.builder().addAllSortFields(fields).input(input).build();
-    if (sort.fetch == null && sort.offset == null) {
-      return convertedSort;
+    Rel input = apply(sort.getInput());
+    Rel output = input;
+
+    // SortRel is added BEFORE FetchRel, if present
+    if (!sort.getCollation().getFieldCollations().isEmpty()) {
+      List<Expression.SortField> fields =
+          sort.getCollation().getFieldCollations().stream()
+              .map(t -> toSortField(t, input.getRecordType()))
+              .collect(java.util.stream.Collectors.toList());
+      output = Sort.builder().addAllSortFields(fields).input(output).build();
     }
-    var offset = Optional.ofNullable(sort.offset).map(r -> asLong(r)).orElse(0L);
-    var builder = Fetch.builder().input(convertedSort).offset(offset);
-    if (sort.fetch == null) {
-      return builder.build();
+
+    if (sort.fetch != null || sort.offset != null) {
+      Long offset = Optional.ofNullable(sort.offset).map(this::asLong).orElse(0L);
+      OptionalLong count =
+          Optional.ofNullable(sort.fetch)
+              .map(r -> OptionalLong.of(asLong(r)))
+              .orElse(OptionalLong.empty());
+
+      var builder = Fetch.builder().input(output).offset(offset).count(count);
+      output = builder.build();
     }
 
-    return builder.count(asLong(sort.fetch)).build();
+    return output;
   }
 
   private long asLong(RexNode rex) {

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

@@ -0,0 +1,34 @@
+package io.substrait.isthmus;
+
+import io.substrait.dsl.SubstraitBuilder;
+import io.substrait.relation.Rel;
+import io.substrait.type.TypeCreator;
+import java.util.List;
+import org.junit.jupiter.api.Test;
+
+public class FetchTest extends PlanTestBase {
+
+  static final TypeCreator R = TypeCreator.of(false);
+
+  final SubstraitBuilder b = new SubstraitBuilder(extensions);
+
+  final Rel TABLE = b.namedScan(List.of("test"), List.of("col1"), List.of(R.STRING));
+
+  @Test
+  void limitOnly() {
+    Rel rel = b.limit(50, TABLE);
+    assertFullRoundTrip(rel);
+  }
+
+  @Test
+  void offsetOnly() {
+    Rel rel = b.offset(50, TABLE);
+    assertFullRoundTrip(rel);
+  }
+
+  @Test
+  void offsetAndLimit() {
+    Rel rel = b.fetch(50, 10, TABLE);
+    assertFullRoundTrip(rel);
+  }
+}