Remove legacy type checking code and update documentation

- Remove PPLTypeChecker, CoercionUtils, CalciteFuncSignature (legacy type checking)
- Simplify LambdaUtils to remove PPLTypeChecker dependency
- Inline datetime widening logic in PPLFuncImpTable (replacing CoercionUtils call)
- Remove OpenSearchSparkSqlDialect (moved to core module)
- Update documentation for new validation pipeline
- Includes all changes from sub-PRs A, B, and C

Author: yuancu

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

@@ -17,13 +17,20 @@
 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.sql.validate.SqlValidator;
 import org.apache.calcite.tools.FrameworkConfig;
 import org.opensearch.sql.ast.expression.UnresolvedExpression;
 import org.opensearch.sql.calcite.utils.CalciteToolsHelper;
 import org.opensearch.sql.calcite.utils.CalciteToolsHelper.OpenSearchRelBuilder;
+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 +79,14 @@ 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;
+
   private CalcitePlanContext(FrameworkConfig config, SysLimit sysLimit, QueryType queryType) {
     this.config = config;
     this.sysLimit = sysLimit;
@@ -101,6 +116,34 @@ private CalcitePlanContext(CalcitePlanContext parent) {
     this.inLambdaContext = true; // Mark that we're inside a lambda
   }
 
+  /**
+   * Creates a new SqlValidator instance. SqlValidator is stateful and should not be reused across
+   * validations, so a new instance is created for each call.
+   *
+   * @return new SqlValidator instance
+   */
+  public SqlValidator getValidator() {
+    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);
+    return PplValidator.create(
+        config, operatorTableProvider.getOperatorTable(), TYPE_FACTORY, validatorConfig);
+  }
+
   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;
@@ -56,10 +59,14 @@
 import org.apache.calcite.rex.RexBuilder;
 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.SqlLibraryOperators;
@@ -765,8 +772,8 @@ public RelNode visitTranspose(
             .map(
                 f ->
                     Map.entry(
-                        ImmutableList.of(rx.makeLiteral(f)),
-                        ImmutableList.of((RexNode) rx.makeCast(varchar, b.field(f), true))))
+                        ImmutableList.of((RexLiteral) rx.makeLiteral(f, varchar, true)),
+                        ImmutableList.of(rx.makeCast(varchar, b.field(f), true, true))))
             .collect(Collectors.toList()));
 
     // Step 3: Trim spaces from columnName column before pivot
@@ -1795,6 +1802,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
@@ -1836,6 +1844,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());
+                      }
+                    }))
+        .collect(Collectors.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

@@ -579,26 +579,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);
+  }
 }