fix(isthmus)!: honor decimal arithmetic overflow options

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

@@ -0,0 +1,81 @@
+package io.substrait.isthmus.expression;
+
+import io.substrait.expression.Expression;
+import io.substrait.expression.FunctionOption;
+import io.substrait.extension.DefaultExtensionCatalog;
+import io.substrait.extension.SimpleExtension.ScalarFunctionVariant;
+import java.util.List;
+import java.util.Locale;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.sql.SqlOperator;
+import org.apache.calcite.sql.type.SqlTypeName;
+
+/** Decimal arithmetic overflow options supported by Calcite. */
+final class DecimalFunctionOptions implements ScalarFunctionOptionPolicy {
+
+  private static SqlOperator operator(ScalarFunctionVariant function) {
+    if (!DefaultExtensionCatalog.FUNCTIONS_ARITHMETIC_DECIMAL.equals(function.urn())
+        || !function.key().equals(function.name() + ":dec_dec")
+        || !List.of("add", "subtract", "multiply", "divide", "modulus").contains(function.name())) {
+      return null;
+    }
+    return FunctionMappings.SCALAR_SIGS.stream()
+        .filter(sig -> sig.name().equals(function.name()))
+        .map(FunctionMappings.Sig::operator)
+        .findFirst()
+        .orElse(null);
+  }
+
+  @Override
+  public List<FunctionOption> forCall(RexCall call, ScalarFunctionVariant function) {
+    SqlOperator operator = operator(function);
+    if (operator == null
+        || call.getOperator() != operator
+        || call.getType().getSqlTypeName() != SqlTypeName.DECIMAL) {
+      return List.of();
+    }
+    // Calcite's BigDecimal arithmetic does not enforce the declared result precision.
+    return List.of(FunctionOption.builder().name("overflow").addValues("SILENT").build());
+  }
+
+  @Override
+  public SqlOperator resolve(Expression.ScalarFunctionInvocation expression, SqlOperator selected) {
+    SqlOperator nativeOperator = operator(expression.declaration());
+    if (nativeOperator == null || expression.options().isEmpty()) {
+      return selected;
+    }
+    if (selected != nativeOperator) {
+      throw new UnsupportedOperationException(
+          "No decimal option policy for Calcite operator " + selected.getName());
+    }
+    // A remainder is bounded by both operands, so valid decimal inputs cannot overflow
+    // the spec's modulus result type. The other operators can exceed its precision.
+    List<String> supported =
+        expression.declaration().name().equals("modulus")
+            ? List.of("SILENT", "SATURATE", "ERROR")
+            : List.of("SILENT");
+    String previous = null;
+    for (FunctionOption option : expression.options()) {
+      ScalarFunctionOptionPolicy.requireDeclaredValues(expression, option);
+      String name = option.getName().toLowerCase(Locale.ROOT);
+      if (!name.equals("overflow")) {
+        throw new UnsupportedOperationException("Unsupported decimal arithmetic option: " + name);
+      }
+      String value =
+          option.values().stream()
+              .map(v -> v.toUpperCase(Locale.ROOT))
+              .filter(supported::contains)
+              .findFirst()
+              .orElseThrow(
+                  () ->
+                      new UnsupportedOperationException(
+                          "Unsupported decimal arithmetic overflow preferences: "
+                              + option.values()));
+      if (previous != null && !previous.equals(value)) {
+        throw new UnsupportedOperationException("Conflicting decimal arithmetic option: overflow");
+      }
+      previous = value;
+    }
+    return selected;
+  }
+}

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

@@ -50,7 +50,8 @@ public class ScalarFunctionConverter
   private final List<ScalarFunctionMapper> mappers;
 
   private final List<ScalarFunctionOptionPolicy> optionPolicies =
-      List.of(new StringFunctionOptions(), new IntegerFunctionOptions());
+      List.of(
+          new StringFunctionOptions(), new IntegerFunctionOptions(), new DecimalFunctionOptions());
 
   /**
    * Creates a converter with the given functions and type factory.