feat(isthmus): observe window function return types

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

@@ -88,6 +88,9 @@ public class ConverterProvider {
   /** The Calcite SQL parser configuration, controlling parsing behaviour like identifier casing. */
   protected final SqlParser.Config sqlParserConfig;
 
+  /** Observer for supplied and independently inferred expression types. */
+  protected final TypeObserver typeObserver;
+
   /** Converter for Substrait scalar functions. */
   protected ScalarFunctionConverter scalarFunctionConverter;
 
@@ -221,6 +224,7 @@ protected ConverterProvider(Builder builder) {
             .unquotedCasing
             .map(builder.sqlParserConfig::withUnquotedCasing)
             .orElse(builder.sqlParserConfig);
+    this.typeObserver = builder.typeObserver;
 
     this.scalarFunctionConverter =
         builder.scalarFunctionConverter.orElseGet(
@@ -429,12 +433,13 @@ public ExpressionRexConverter getExpressionRexConverter(
   /**
    * Returns the observer for supplied and independently inferred expression types.
    *
-   * <p>Override to collect type observations during Substrait-to-Calcite conversion.
+   * <p>Configure via {@link Builder#typeObserver(TypeObserver)} or override this method to collect
+   * type observations during Substrait-to-Calcite conversion.
    *
    * @return a no-op observer by default
    */
   public TypeObserver getTypeObserver() {
-    return TypeObserver.NOOP;
+    return typeObserver;
   }
 
   /**
@@ -565,6 +570,7 @@ public static class Builder {
     private TypeConverter typeConverter = TypeConverter.DEFAULT;
     private Plan.ExecutionBehavior executionBehavior = createDefaultExecutionBehavior();
     private SqlParser.Config sqlParserConfig = DEFAULT_SQL_PARSER_CONFIG;
+    private TypeObserver typeObserver = TypeObserver.NOOP;
     private Optional<Casing> unquotedCasing = Optional.empty();
 
     // Derived from the extensions and type factory at build time when left unset.
@@ -646,6 +652,17 @@ public Builder unquotedCasing(Casing unquotedCasing) {
       return this;
     }
 
+    /**
+     * Sets the observer for supplied and independently inferred expression types.
+     *
+     * @param typeObserver the type observer
+     * @return this builder
+     */
+    public Builder typeObserver(TypeObserver typeObserver) {
+      this.typeObserver = typeObserver;
+      return this;
+    }
+
     /**
      * Sets the scalar function converter. When left unset, it is derived from the configured
      * extensions and type factory.

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

@@ -39,6 +39,7 @@
 import org.apache.calcite.rel.type.RelDataType;
 import org.apache.calcite.rel.type.RelDataTypeFactory;
 import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexCallBinding;
 import org.apache.calcite.rex.RexFieldCollation;
 import org.apache.calcite.rex.RexInputRef;
 import org.apache.calcite.rex.RexLambdaRef;
@@ -51,6 +52,7 @@
 import org.apache.calcite.sql.SqlIntervalQualifier;
 import org.apache.calcite.sql.SqlKind;
 import org.apache.calcite.sql.SqlOperator;
+import org.apache.calcite.sql.SqlWindow;
 import org.apache.calcite.sql.fun.SqlStdOperatorTable;
 import org.apache.calcite.sql.parser.SqlParserPos;
 import org.apache.calcite.sql.type.SqlTypeName;
@@ -549,31 +551,32 @@ public RexNode visit(Expression.ScalarFunctionInvocation expr, Context context)
 
     RelDataType returnType = typeConverter.toCalcite(typeFactory, expr.outputType());
     RexNode rexCall = rexBuilder.makeCall(returnType, operator, args);
-    // If type observations are not needed, avoid recomputing the RexCall with Calcite's
-    // independently inferred return type.
-    if (typeObserver == TypeObserver.NOOP) {
-      return rexCall;
-    }
-    observeScalarType(expr, () -> rexBuilder.makeCall(operator, args));
+    observeType(
+        expr,
+        TypeObservation.Source.SCALAR_FUNCTION,
+        () -> rexBuilder.deriveReturnType(operator, args));
     return rexCall;
   }
 
-  private void observeScalarType(
-      Expression.ScalarFunctionInvocation expression, Supplier<RexNode> inferredCallSupplier) {
+  private void observeType(
+      Expression expression,
+      TypeObservation.Source source,
+      Supplier<RelDataType> inferredTypeSupplier) {
+    // The conversion result never depends on the independently inferred type, so when no observer
+    // is installed skip Calcite's inference entirely.
+    if (typeObserver == TypeObserver.NOOP) {
+      return;
+    }
     TypeObservation observation;
-    RexNode inferredCall;
+    RelDataType inferredType;
     try {
-      inferredCall = inferredCallSupplier.get();
+      inferredType = inferredTypeSupplier.get();
     } catch (RuntimeException inferenceFailure) {
-      observation =
-          TypeObservation.failure(
-              TypeObservation.Source.SCALAR_FUNCTION, expression, inferenceFailure);
+      observation = TypeObservation.failure(source, expression, inferenceFailure);
       typeObserver.observe(observation);
       return;
     }
-    observation =
-        TypeObservation.success(
-            TypeObservation.Source.SCALAR_FUNCTION, expression, inferredCall.getType());
+    observation = TypeObservation.success(source, expression, inferredType);
     typeObserver.observe(observation);
   }
 
@@ -631,19 +634,52 @@ public RexNode visit(Expression.WindowFunctionInvocation expr, Context context)
     boolean nullWhenCountZero = false;
     boolean allowPartial = true;
 
-    return rexBuilder.makeOver(
-        outputType,
-        (SqlAggFunction) operator,
-        args,
-        partitionKeys,
-        orderKeys,
-        lowerBound,
-        upperBound,
-        rowMode,
-        allowPartial,
-        nullWhenCountZero,
-        distinct,
-        ignoreNulls);
+    RexNode rexOver =
+        rexBuilder.makeOver(
+            outputType,
+            (SqlAggFunction) operator,
+            args,
+            partitionKeys,
+            orderKeys,
+            lowerBound,
+            upperBound,
+            rowMode,
+            allowPartial,
+            nullWhenCountZero,
+            distinct,
+            ignoreNulls);
+    observeType(
+        expr,
+        TypeObservation.Source.WINDOW_FUNCTION,
+        () -> operator.inferReturnType(windowBinding(operator, args, lowerBound, upperBound)));
+    return rexOver;
+  }
+
+  /**
+   * Builds the operand binding Calcite itself uses when inferring the return type of a windowed
+   * aggregate.
+   *
+   * <p>{@link RexBuilder#deriveReturnType(SqlOperator, java.util.List)} binds the operands with a
+   * plain {@link RexCallBinding}, whose {@link
+   * org.apache.calcite.sql.SqlOperatorBinding#hasEmptyGroup()} is always {@code false}. Calcite's
+   * validator instead derives that flag from the window bounds, so any return type strategy that
+   * consults it (for example {@code ARG0_NULLABLE_IF_EMPTY} or {@code AGG_SUM}) widens its result
+   * to nullable over a frame that may be empty. Binding without the flag would report a
+   * non-nullable inferred type for such an operator, so a genuine nullability deviation would be
+   * observed as a match.
+   */
+  private RexCallBinding windowBinding(
+      SqlOperator operator,
+      List<RexNode> operands,
+      RexWindowBound lowerBound,
+      RexWindowBound upperBound) {
+    boolean emptyGroup = !SqlWindow.isAlwaysNonEmpty(lowerBound, upperBound);
+    return new RexCallBinding(typeFactory, operator, operands, ImmutableList.of()) {
+      @Override
+      public boolean hasEmptyGroup() {
+        return emptyGroup;
+      }
+    };
   }
 
   private Set<SqlKind> asSqlKind(Expression.SortDirection direction) {

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

@@ -15,7 +15,10 @@ public final class TypeObservation {
   /** The expression category that produced an observation. */
   public enum Source {
     /** A scalar function invocation. */
-    SCALAR_FUNCTION
+    SCALAR_FUNCTION,
+
+    /** A window function invocation. */
+    WINDOW_FUNCTION
   }
 
   private final Source source;

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

@@ -11,6 +11,7 @@
 import io.substrait.extension.SimpleExtension;
 import io.substrait.isthmus.expression.AggregateFunctionConverter;
 import io.substrait.isthmus.expression.ScalarFunctionConverter;
+import io.substrait.isthmus.expression.TypeObserver;
 import io.substrait.isthmus.expression.WindowFunctionConverter;
 import org.apache.calcite.avatica.util.Casing;
 import org.apache.calcite.rel.type.RelDataTypeFactory;
@@ -33,6 +34,7 @@ void derivesFunctionConvertersWhenUnset() {
     assertNotNull(provider.getAggregateFunctionConverter());
     assertNotNull(provider.getWindowFunctionConverter());
     assertEquals(ConverterProvider.DEFAULT_SQL_PARSER_CONFIG, provider.getSqlParserConfig());
+    assertSame(TypeObserver.NOOP, provider.getTypeObserver());
   }
 
   @Test