Use Calcite's validation system for type checking & coercion

api/src/test/java/org/opensearch/sql/api/transpiler/UnifiedQueryTranspilerTest.java

@@ -12,7 +12,7 @@
 import org.junit.Before;
 import org.junit.Test;
 import org.opensearch.sql.api.UnifiedQueryTestBase;
-import org.opensearch.sql.ppl.calcite.OpenSearchSparkSqlDialect;
+import org.opensearch.sql.calcite.validate.OpenSearchSparkSqlDialect;
 
 public class UnifiedQueryTranspilerTest extends UnifiedQueryTestBase {
 
@@ -43,11 +43,9 @@ public void testToSqlWithCustomDialect() {
     UnifiedQueryTranspiler customTranspiler =
         UnifiedQueryTranspiler.builder().dialect(OpenSearchSparkSqlDialect.DEFAULT).build();
     String actualSql = customTranspiler.toSql(plan);
-    String expectedSql =
-        normalize(
-            "SELECT *\nFROM `catalog`.`employees`\nWHERE TRY_CAST(`name` AS DOUBLE) = 1.230E2");
+    String expectedSql = normalize("SELECT *\nFROM `catalog`.`employees`\nWHERE `name` = 123");
     assertEquals(
-        "Transpiled query using OpenSearchSparkSqlDialect should translate SAFE_CAST to TRY_CAST",
+        "Numeric types can be implicitly coerced to string with OpenSearchSparkSqlDialect",
         expectedSql,
         actualSql);
   }

core/src/main/java/org/opensearch/sql/calcite/CalcitePlanContext.java

@@ -8,6 +8,7 @@
 import static org.opensearch.sql.calcite.utils.OpenSearchTypeFactory.TYPE_FACTORY;
 
 import java.sql.Connection;
+import java.sql.SQLException;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
@@ -17,13 +18,21 @@
 import java.util.function.BiFunction;
 import lombok.Getter;
 import lombok.Setter;
+import org.apache.calcite.config.NullCollation;
 import org.apache.calcite.rex.RexCorrelVariable;
 import org.apache.calcite.rex.RexLambdaRef;
 import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.server.CalciteServerStatement;
+import org.apache.calcite.sql.validate.SqlValidator;
 import org.apache.calcite.tools.FrameworkConfig;
 import org.apache.calcite.tools.RelBuilder;
 import org.opensearch.sql.ast.expression.UnresolvedExpression;
 import org.opensearch.sql.calcite.utils.CalciteToolsHelper;
+import org.opensearch.sql.calcite.validate.OpenSearchSparkSqlDialect;
+import org.opensearch.sql.calcite.validate.PplTypeCoercion;
+import org.opensearch.sql.calcite.validate.PplTypeCoercionRule;
+import org.opensearch.sql.calcite.validate.PplValidator;
+import org.opensearch.sql.calcite.validate.SqlOperatorTableProvider;
 import org.opensearch.sql.common.setting.Settings;
 import org.opensearch.sql.executor.QueryType;
 import org.opensearch.sql.expression.function.FunctionProperties;
@@ -72,6 +81,17 @@ public class CalcitePlanContext {
   /** Whether we're currently inside a lambda context. */
   @Getter @Setter private boolean inLambdaContext = false;
 
+  /**
+   * -- SETTER -- Sets the SQL operator table provider. This must be called during initialization by
+   * the opensearch module.
+   *
+   * @param provider the provider to use for obtaining operator tables
+   */
+  @Setter private static SqlOperatorTableProvider operatorTableProvider;
+
+  /** Cached SqlValidator instance (lazy initialized). */
+  private volatile SqlValidator validator;
+
   private CalcitePlanContext(FrameworkConfig config, SysLimit sysLimit, QueryType queryType) {
     this.config = config;
     this.sysLimit = sysLimit;
@@ -101,6 +121,52 @@ private CalcitePlanContext(CalcitePlanContext parent) {
     this.inLambdaContext = true; // Mark that we're inside a lambda
   }
 
+  /**
+   * Gets the SqlValidator instance (singleton per CalcitePlanContext).
+   *
+   * @return cached SqlValidator instance
+   */
+  public SqlValidator getValidator() {
+    if (validator == null) {
+      synchronized (this) {
+        //  Double-Checked Locking for thread-safety
+        if (validator == null) {
+          final CalciteServerStatement statement;
+          try {
+            statement = connection.createStatement().unwrap(CalciteServerStatement.class);
+          } catch (SQLException e) {
+            throw new RuntimeException(e);
+          }
+          if (operatorTableProvider == null) {
+            throw new IllegalStateException(
+                "SqlOperatorTableProvider must be set before creating CalcitePlanContext");
+          }
+          SqlValidator.Config validatorConfig =
+              SqlValidator.Config.DEFAULT
+                  .withTypeCoercionRules(PplTypeCoercionRule.instance())
+                  .withTypeCoercionFactory(PplTypeCoercion::create)
+                  // Use lenient conformance for PPL compatibility
+                  .withConformance(OpenSearchSparkSqlDialect.DEFAULT.getConformance())
+                  // Use Spark SQL's NULL collation (NULLs sorted LOW/FIRST)
+                  .withDefaultNullCollation(NullCollation.LOW)
+                  // This ensures that coerced arguments are replaced with cast version in sql
+                  // select list because coercion is performed during select list expansion during
+                  // sql validation. Affects 4356.yml
+                  // See SqlValidatorImpl#validateSelectList and AggConverter#translateAgg
+                  .withIdentifierExpansion(true);
+          validator =
+              PplValidator.create(
+                  statement,
+                  config,
+                  operatorTableProvider.getOperatorTable(),
+                  TYPE_FACTORY,
+                  validatorConfig);
+        }
+      }
+    }
+    return validator;
+  }
+
   public RexNode resolveJoinCondition(
       UnresolvedExpression expr,
       BiFunction<UnresolvedExpression, CalcitePlanContext, RexNode> transformFunction) {

core/src/main/java/org/opensearch/sql/calcite/CalciteRelNodeVisitor.java

@@ -43,9 +43,12 @@
 import java.util.stream.IntStream;
 import java.util.stream.Stream;
 import lombok.AllArgsConstructor;
+import lombok.NonNull;
 import org.apache.calcite.adapter.enumerable.RexToLixTranslator;
 import org.apache.calcite.plan.RelOptTable;
 import org.apache.calcite.plan.ViewExpanders;
+import org.apache.calcite.rel.RelCollation;
+import org.apache.calcite.rel.RelFieldCollation;
 import org.apache.calcite.rel.RelNode;
 import org.apache.calcite.rel.core.Aggregate;
 import org.apache.calcite.rel.core.JoinRelType;
@@ -55,10 +58,14 @@
 import org.apache.calcite.rel.type.RelDataTypeField;
 import org.apache.calcite.rex.RexCall;
 import org.apache.calcite.rex.RexCorrelVariable;
+import org.apache.calcite.rex.RexFieldCollation;
 import org.apache.calcite.rex.RexInputRef;
 import org.apache.calcite.rex.RexLiteral;
 import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexOver;
+import org.apache.calcite.rex.RexShuttle;
 import org.apache.calcite.rex.RexVisitorImpl;
+import org.apache.calcite.rex.RexWindow;
 import org.apache.calcite.rex.RexWindowBounds;
 import org.apache.calcite.sql.SqlKind;
 import org.apache.calcite.sql.fun.SqlStdOperatorTable;
@@ -1740,6 +1747,7 @@ public RelNode visitStreamWindow(StreamWindow node, CalcitePlanContext context)
     // Default: first get rawExpr
     List<RexNode> overExpressions =
         node.getWindowFunctionList().stream().map(w -> rexVisitor.analyze(w, context)).toList();
+    overExpressions = embedExistingCollationsIntoOver(overExpressions, context);
 
     if (hasGroup) {
       // only build sequence when there is by condition
@@ -1781,6 +1789,84 @@ public RelNode visitStreamWindow(StreamWindow node, CalcitePlanContext context)
     return context.relBuilder.peek();
   }
 
+  /**
+   * Embed existing collation into window function's over clauses.
+   *
+   * <p>Window functions with frame specifications like {@code ROWS n PRECEDING} require ORDER BY to
+   * determine row order. Without it, results are non-deterministic.
+   *
+   * <p>Without this fix, the initial plan has ORDER BY separate from window functions:
+   *
+   * <pre>
+   * LogicalProject(SUM($5) OVER (ROWS 1 PRECEDING))  ← Missing ORDER BY
+   *   LogicalSort(sort0=[$5])
+   * </pre>
+   *
+   * <p>This causes problems during validation as the order is not bound to the window. With this
+   * fix, sort collations are embeded into each {@code RexOver} window:
+   *
+   * <pre>
+   * LogicalProject(SUM($5) OVER (ORDER BY $5 ROWS 1 PRECEDING))  ← ORDER BY embedded
+   * </pre>
+   *
+   * @param overExpressions Window function expressions (may contain nested {@link RexOver})
+   * @param context Plan context for building RexNodes
+   * @return Expressions with ORDER BY embedded in all window specifications
+   */
+  private List<RexNode> embedExistingCollationsIntoOver(
+      List<RexNode> overExpressions, CalcitePlanContext context) {
+    RelCollation existingCollation = context.relBuilder.peek().getTraitSet().getCollation();
+    List<@NonNull RelFieldCollation> relCollations =
+        existingCollation == null ? List.of() : existingCollation.getFieldCollations();
+    ImmutableList<@NonNull RexFieldCollation> rexCollations =
+        relCollations.stream()
+            .map(f -> relCollationToRexCollation(f, context.relBuilder))
+            .collect(ImmutableList.toImmutableList());
+    return overExpressions.stream()
+        .map(
+            n ->
+                n.accept(
+                    new RexShuttle() {
+                      @Override
+                      public RexNode visitOver(RexOver over) {
+                        RexWindow window = over.getWindow();
+                        return context.rexBuilder.makeOver(
+                            over.getType(),
+                            over.getAggOperator(),
+                            over.getOperands(),
+                            window.partitionKeys,
+                            rexCollations,
+                            window.getLowerBound(),
+                            window.getUpperBound(),
+                            window.isRows(),
+                            true,
+                            false,
+                            over.isDistinct(),
+                            over.ignoreNulls());
+                      }
+                    }))
+        .toList();
+  }
+
+  private static RexFieldCollation relCollationToRexCollation(
+      RelFieldCollation relCollation, RelBuilder builder) {
+    RexNode fieldRef = builder.field(relCollation.getFieldIndex());
+
+    // Convert direction flags to SqlKind set
+    Set<SqlKind> flags = new HashSet<>();
+    if (relCollation.direction == RelFieldCollation.Direction.DESCENDING
+        || relCollation.direction == RelFieldCollation.Direction.STRICTLY_DESCENDING) {
+      flags.add(SqlKind.DESCENDING);
+    }
+    if (relCollation.nullDirection == RelFieldCollation.NullDirection.FIRST) {
+      flags.add(SqlKind.NULLS_FIRST);
+    } else if (relCollation.nullDirection == RelFieldCollation.NullDirection.LAST) {
+      flags.add(SqlKind.NULLS_LAST);
+    }
+
+    return new RexFieldCollation(fieldRef, flags);
+  }
+
   private List<RexNode> wrapWindowFunctionsWithGroupNotNull(
       List<RexNode> overExpressions, RexNode groupNotNull, CalcitePlanContext context) {
     List<RexNode> wrappedOverExprs = new ArrayList<>(overExpressions.size());

core/src/main/java/org/opensearch/sql/calcite/CalciteRexNodeVisitor.java

@@ -480,26 +480,8 @@ public RexNode visitWindowFunction(WindowFunction node, CalcitePlanContext conte
                   (arguments.isEmpty() || arguments.size() == 1)
                       ? Collections.emptyList()
                       : arguments.subList(1, arguments.size());
-              List<RexNode> nodes =
-                  PPLFuncImpTable.INSTANCE.validateAggFunctionSignature(
-                      functionName, field, args, context.rexBuilder);
-              return nodes != null
-                  ? PlanUtils.makeOver(
-                      context,
-                      functionName,
-                      nodes.getFirst(),
-                      nodes.size() <= 1 ? Collections.emptyList() : nodes.subList(1, nodes.size()),
-                      partitions,
-                      List.of(),
-                      node.getWindowFrame())
-                  : PlanUtils.makeOver(
-                      context,
-                      functionName,
-                      field,
-                      args,
-                      partitions,
-                      List.of(),
-                      node.getWindowFrame());
+              return PlanUtils.makeOver(
+                  context, functionName, field, args, partitions, List.of(), node.getWindowFrame());
             })
         .orElseThrow(
             () ->

core/src/main/java/org/opensearch/sql/calcite/ExtendedRexBuilder.java

@@ -14,14 +14,19 @@
 import org.apache.calcite.rex.RexBuilder;
 import org.apache.calcite.rex.RexLiteral;
 import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.sql.SqlCallBinding;
 import org.apache.calcite.sql.SqlIntervalQualifier;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlOperator;
 import org.apache.calcite.sql.fun.SqlStdOperatorTable;
 import org.apache.calcite.sql.parser.SqlParserPos;
 import org.apache.calcite.sql.type.SqlTypeName;
 import org.apache.calcite.sql.type.SqlTypeUtil;
+import org.apache.calcite.sql.validate.implicit.TypeCoercionImpl;
 import org.opensearch.sql.ast.expression.SpanUnit;
 import org.opensearch.sql.calcite.type.AbstractExprRelDataType;
 import org.opensearch.sql.calcite.utils.OpenSearchTypeFactory;
+import org.opensearch.sql.calcite.utils.OpenSearchTypeUtil;
 import org.opensearch.sql.data.type.ExprCoreType;
 import org.opensearch.sql.exception.ExpressionEvaluationException;
 import org.opensearch.sql.exception.SemanticCheckException;
@@ -146,7 +151,7 @@ public RexNode makeCast(
         //            SqlStdOperatorTable.NOT_EQUALS,
         //            ImmutableList.of(exp, makeZeroLiteral(sourceType)));
       }
-    } else if (OpenSearchTypeFactory.isUserDefinedType(type)) {
+    } else if (OpenSearchTypeUtil.isUserDefinedType(type)) {
       if (RexLiteral.isNullLiteral(exp)) {
         return super.makeCast(pos, type, exp, matchNullability, safe, format);
       }
@@ -185,4 +190,33 @@ else if ((SqlTypeUtil.isApproximateNumeric(sourceType) || SqlTypeUtil.isDecimal(
     }
     return super.makeCast(pos, type, exp, matchNullability, safe, format);
   }
+
+  /**
+   * Derives the return type of call to an operator.
+   *
+   * <p>In Calcite, coercion between STRING and NUMERIC operands takes place during converting SQL
+   * to RelNode. However, as we are building logical plans directly, the coercion is not yet
+   * implemented at this point. Hence, we duplicate {@link
+   * TypeCoercionImpl#binaryArithmeticWithStrings} here to infer the correct type, enabling
+   * operations like {@code "5" / 10}. The actual coercion will be inserted later when performing
+   * validation on SqlNode.
+   *
+   * @see TypeCoercionImpl#binaryArithmeticCoercion(SqlCallBinding)
+   * @param op the operator being called
+   * @param exprs actual operands
+   * @return derived type
+   */
+  @Override
+  public RelDataType deriveReturnType(SqlOperator op, List<? extends RexNode> exprs) {
+    if (op.getKind().belongsTo(SqlKind.BINARY_ARITHMETIC) && exprs.size() == 2) {
+      final RelDataType type1 = exprs.get(0).getType();
+      final RelDataType type2 = exprs.get(1).getType();
+      if (SqlTypeUtil.isNumeric(type1) && OpenSearchTypeUtil.isCharacter(type2)) {
+        return type1;
+      } else if (OpenSearchTypeUtil.isCharacter(type1) && SqlTypeUtil.isNumeric(type2)) {
+        return type2;
+      }
+    }
+    return super.deriveReturnType(op, exprs);
+  }
 }

core/src/main/java/org/opensearch/sql/calcite/utils/CalciteToolsHelper.java

@@ -240,7 +240,7 @@ private void registerCustomizedRules(RelOptPlanner planner) {
      * return {@link OpenSearchCalcitePreparingStmt}
      */
     @Override
-    protected CalcitePrepareImpl.CalcitePreparingStmt getPreparingStmt(
+    public CalcitePrepareImpl.CalcitePreparingStmt getPreparingStmt(
         CalcitePrepare.Context context,
         Type elementType,
         CalciteCatalogReader catalogReader,
@@ -369,7 +369,8 @@ public RelNode visit(TableScan scan) {
               "The 'bins' parameter on timestamp fields requires: (1) pushdown to be enabled"
                   + " (controlled by plugins.calcite.pushdown.enabled, enabled by default), and"
                   + " (2) the timestamp field to be used as an aggregation bucket (e.g., 'stats"
-                  + " count() by @timestamp').");
+                  + " count() by @timestamp').",
+              e);
         }
         throw Util.throwAsRuntime(e);
       }