fix(isthmus)!: translate array indexes correctly

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

@@ -12,6 +12,9 @@
 import org.apache.calcite.rex.RexLiteral;
 import org.apache.calcite.rex.RexNode;
 import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlOperator;
+import org.apache.calcite.sql.fun.SqlItemOperator;
+import org.apache.calcite.sql.type.SqlTypeName;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -43,7 +46,7 @@ public FieldSelectionConverter(TypeConverter typeConverter) {
    *
    * <ul>
    *   <li>ROW dereference by integer index
-   *   <li>ARRAY dereference by integer index
+   *   <li>ARRAY dereference by integer index, translated to Substrait's zero-based offset
    *   <li>MAP dereference by string key
    * </ul>
    *
@@ -65,19 +68,26 @@ public Optional<Expression> convert(
       LOGGER
           .atWarn()
           .log(
-              "Found item operator without literal kind/type. This isn't handled well. Reference was {} with toString {}.",
+              "Found item operator without literal kind/type. This isn't handled well. Reference"
+                  + " was {} with toString {}.",
               reference.getKind().name(),
               reference);
       return Optional.empty();
     }
 
-    Literal literal = (new LiteralConverter(typeConverter)).convert((RexLiteral) reference);
+    SqlTypeName containerType = toDereference.getType().getSqlTypeName();
+    // A Substrait list reference returns null for an out-of-range index, so it cannot preserve
+    // the error behavior of OFFSET or ORDINAL. Decline before the array operand is converted.
+    if (containerType == SqlTypeName.ARRAY && !isSafeArrayOperator(call.getOperator())) {
+      return Optional.empty();
+    }
 
     Expression input = topLevelConverter.apply(toDereference);
 
-    switch (toDereference.getType().getSqlTypeName()) {
+    switch (containerType) {
       case ROW:
         {
+          Literal literal = (new LiteralConverter(typeConverter)).convert((RexLiteral) reference);
           Optional<Integer> index = toInt(literal);
           if (index.isEmpty()) {
             return Optional.empty();
@@ -90,20 +100,33 @@ public Optional<Expression> convert(
         }
       case ARRAY:
         {
-          Optional<Integer> index = toInt(literal);
-          if (index.isEmpty()) {
-            return Optional.empty();
+          SqlItemOperator operator = (SqlItemOperator) call.getOperator();
+          long offset = Integer.MAX_VALUE;
+          if (!((RexLiteral) reference).isNull()) {
+            Literal literal = (new LiteralConverter(typeConverter)).convert((RexLiteral) reference);
+            Optional<Long> index = toLong(literal);
+            if (index.isEmpty()) {
+              return Optional.empty();
+            }
+            // Negative Substrait offsets count from the end, whereas Calcite returns null below
+            // the operator's base. Substrait lists have at most Integer.MAX_VALUE elements, so that
+            // zero-based offset cannot select an element. Keep the reference to evaluate the
+            // array operand and preserve its element type, including for a null index.
+            if (index.get() >= operator.offset) {
+              offset = Math.min(index.get() - operator.offset, Integer.MAX_VALUE);
+            }
           }
 
           if (input instanceof FieldReference) {
-            return Optional.of(((FieldReference) input).dereferenceList(index.get()));
+            return Optional.of(((FieldReference) input).dereferenceList((int) offset));
           } else {
-            return Optional.of(FieldReference.newListReference(index.get(), input));
+            return Optional.of(FieldReference.newListReference((int) offset, input));
           }
         }
 
       case MAP:
         {
+          Literal literal = (new LiteralConverter(typeConverter)).convert((RexLiteral) reference);
           Optional<String> mapKey = toString(literal);
           if (mapKey.isEmpty()) {
             return Optional.empty();
@@ -121,6 +144,14 @@ public Optional<Expression> convert(
     return Optional.empty();
   }
 
+  private static boolean isSafeArrayOperator(SqlOperator operator) {
+    if (!(operator instanceof SqlItemOperator)) {
+      return false;
+    }
+    SqlItemOperator itemOperator = (SqlItemOperator) operator;
+    return itemOperator.safe && (itemOperator.offset == 0 || itemOperator.offset == 1);
+  }
+
   /**
    * Converts a numeric literal to an integer index.
    *
@@ -141,6 +172,13 @@ private Optional<Integer> toInt(Expression.Literal l) {
     return Optional.empty();
   }
 
+  private Optional<Long> toLong(Expression.Literal literal) {
+    if (literal instanceof Expression.I64Literal) {
+      return Optional.of(((Expression.I64Literal) literal).value());
+    }
+    return toInt(literal).map(Integer::longValue);
+  }
+
   /**
    * Converts a fixed-char literal to a string key.
    *

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

@@ -204,7 +204,7 @@ public RelDataType getRowType(RelDataTypeFactory factory) {
             + "      field: 1 # a\n"
             + "      child {\n"
             + "        list_element {\n"
-            + "          offset: 1\n"
+            + "          offset: 0\n"
             + "        }\n"
             + "      }\n"
             + "    }\n"
@@ -244,13 +244,13 @@ public RelDataType getRowType(RelDataTypeFactory factory) {
             + "      field: 1 # a\n"
             + "      child {\n"
             + "        list_element {\n"
-            + "          offset: 1\n"
+            + "          offset: 0\n"
             + "          child {\n"
             + "            list_element {\n"
-            + "              offset: 2\n"
+            + "              offset: 1\n"
             + "              child {\n"
             + "                list_element {\n"
-            + "                  offset: 3\n"
+            + "                  offset: 2\n"
             + "                }\n"
             + "              }\n"
             + "            }\n"
@@ -300,7 +300,7 @@ public RelDataType getRowType(RelDataTypeFactory factory) {
             + "          field: 0 # .b\n"
             + "          child {\n"
             + "            list_element {\n"
-            + "              offset: 2\n"
+            + "              offset: 1\n"
             + "              child {\n"
             + "                struct_field {\n"
             + "                  field: 0 # .c\n"