Fix binning udf resolution / Add type coercion support for binning UDFs

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

@@ -338,12 +338,12 @@ public static boolean isUserDefinedType(RelDataType type) {
   }
 
   /**
-   * Checks if the RelDataType represents a numeric field. Supports both standard SQL numeric types
-   * (INTEGER, BIGINT, SMALLINT, TINYINT, FLOAT, DOUBLE, DECIMAL, REAL) and OpenSearch UDT numeric
-   * types.
+   * Checks if the RelDataType represents a numeric type. Supports standard SQL numeric types
+   * (INTEGER, BIGINT, SMALLINT, TINYINT, FLOAT, DOUBLE, DECIMAL, REAL), OpenSearch UDT numeric
+   * types, and string types (VARCHAR, CHAR).
    *
    * @param fieldType the RelDataType to check
-   * @return true if the type is numeric, false otherwise
+   * @return true if the type is numeric or string, false otherwise
    */
   public static boolean isNumericType(RelDataType fieldType) {
     // Check standard SQL numeric types
@@ -359,6 +359,11 @@ public static boolean isNumericType(RelDataType fieldType) {
       return true;
     }
 
+    // Check string types (VARCHAR, CHAR)
+    if (sqlType == SqlTypeName.VARCHAR || sqlType == SqlTypeName.CHAR) {
+      return true;
+    }
+
     // Check for OpenSearch UDT numeric types
     if (isUserDefinedType(fieldType)) {
       AbstractExprRelDataType<?> exprType = (AbstractExprRelDataType<?>) fieldType;

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

@@ -134,6 +134,65 @@ private PPLOperandTypes() {}
               SqlTypeFamily.NUMERIC,
               SqlTypeFamily.NUMERIC,
               SqlTypeFamily.NUMERIC));
+
+  public static final UDFOperandMetadata WIDTH_BUCKET_OPERAND =
+      UDFOperandMetadata.wrap(
+          (CompositeOperandTypeChecker)
+              // 1. Numeric fields: bin age span=10
+              OperandTypes.family(
+                      SqlTypeFamily.NUMERIC,
+                      SqlTypeFamily.INTEGER,
+                      SqlTypeFamily.NUMERIC,
+                      SqlTypeFamily.NUMERIC)
+                  // 2. Timestamp fields with OpenSearch type system
+                  // Used in: Production + Integration tests (CalciteBinCommandIT)
+                  .or(
+                      OperandTypes.family(
+                          SqlTypeFamily.TIMESTAMP,
+                          SqlTypeFamily.INTEGER,
+                          SqlTypeFamily.CHARACTER, // TIMESTAMP - TIMESTAMP = INTERVAL (as STRING)
+                          SqlTypeFamily.TIMESTAMP))
+                  // 3. Timestamp fields with Calcite SCOTT schema
+                  // Used in: Unit tests (CalcitePPLBinTest)
+                  .or(
+                      OperandTypes.family(
+                          SqlTypeFamily.TIMESTAMP,
+                          SqlTypeFamily.INTEGER,
+                          SqlTypeFamily.TIMESTAMP, // TIMESTAMP - TIMESTAMP = TIMESTAMP
+                          SqlTypeFamily.TIMESTAMP))
+                  // DATE field with OpenSearch type system
+                  // Used in: Production + Integration tests (CalciteBinCommandIT)
+                  .or(
+                      OperandTypes.family(
+                          SqlTypeFamily.DATE,
+                          SqlTypeFamily.INTEGER,
+                          SqlTypeFamily.CHARACTER, // DATE - DATE = INTERVAL (as STRING)
+                          SqlTypeFamily.DATE))
+                  // DATE field with Calcite SCOTT schema
+                  // Used in: Unit tests (CalcitePPLBinTest)
+                  .or(
+                      OperandTypes.family(
+                          SqlTypeFamily.DATE,
+                          SqlTypeFamily.INTEGER,
+                          SqlTypeFamily.DATE, // DATE - DATE = DATE
+                          SqlTypeFamily.DATE))
+                  // TIME field with OpenSearch type system
+                  // Used in: Production + Integration tests (CalciteBinCommandIT)
+                  .or(
+                      OperandTypes.family(
+                          SqlTypeFamily.TIME,
+                          SqlTypeFamily.INTEGER,
+                          SqlTypeFamily.CHARACTER, // TIME - TIME = INTERVAL (as STRING)
+                          SqlTypeFamily.TIME))
+                  // TIME field with Calcite SCOTT schema
+                  // Used in: Unit tests (CalcitePPLBinTest)
+                  .or(
+                      OperandTypes.family(
+                          SqlTypeFamily.TIME,
+                          SqlTypeFamily.INTEGER,
+                          SqlTypeFamily.TIME, // TIME - TIME = TIME
+                          SqlTypeFamily.TIME)));
+
   public static final UDFOperandMetadata NUMERIC_NUMERIC_NUMERIC_NUMERIC_NUMERIC =
       UDFOperandMetadata.wrap(
           OperandTypes.family(

core/src/main/java/org/opensearch/sql/calcite/utils/binning/BinnableField.java

@@ -11,13 +11,7 @@
 import org.opensearch.sql.calcite.utils.OpenSearchTypeFactory;
 import org.opensearch.sql.exception.SemanticCheckException;
 
-/**
- * Represents a validated field that supports binning operations. The existence of this class
- * guarantees that validation has been run - the field is either numeric or time-based.
- *
- * <p>This design encodes validation in the type system, preventing downstream code from forgetting
- * to validate or running validation multiple times.
- */
+/** Represents a field that supports binning operations. */
 @Getter
 public class BinnableField {
   private final RexNode fieldExpr;
@@ -27,13 +21,12 @@ public class BinnableField {
   private final boolean isNumeric;
 
   /**
-   * Creates a validated BinnableField. Throws SemanticCheckException if the field is neither
-   * numeric nor time-based.
+   * Creates a BinnableField. Validates that the field type is compatible with binning operations.
    *
    * @param fieldExpr The Rex expression for the field
    * @param fieldType The relational data type of the field
    * @param fieldName The name of the field (for error messages)
-   * @throws SemanticCheckException if the field is neither numeric nor time-based
+   * @throws SemanticCheckException if the field type is not supported for binning
    */
   public BinnableField(RexNode fieldExpr, RelDataType fieldType, String fieldName) {
     this.fieldExpr = fieldExpr;
@@ -43,13 +36,10 @@ public BinnableField(RexNode fieldExpr, RelDataType fieldType, String fieldName)
     this.isTimeBased = OpenSearchTypeFactory.isTimeBasedType(fieldType);
     this.isNumeric = OpenSearchTypeFactory.isNumericType(fieldType);
 
-    // Validation: field must be either numeric or time-based
+    // Reject truly unsupported types (e.g., BOOLEAN, ARRAY, MAP)
     if (!isNumeric && !isTimeBased) {
       throw new SemanticCheckException(
-          String.format(
-              "Cannot apply binning: field '%s' is non-numeric and not time-related, expected"
-                  + " numeric or time-related type",
-              fieldName));
+          String.format("Cannot apply binning to field '%s': unsupported type", fieldName));
     }
   }
 

core/src/main/java/org/opensearch/sql/calcite/utils/binning/handlers/CountBinHandler.java

@@ -6,7 +6,6 @@
 package org.opensearch.sql.calcite.utils.binning.handlers;
 
 import org.apache.calcite.rex.RexNode;
-import org.apache.calcite.sql.fun.SqlStdOperatorTable;
 import org.opensearch.sql.ast.expression.Literal;
 import org.opensearch.sql.ast.tree.Bin;
 import org.opensearch.sql.ast.tree.CountBin;
@@ -16,7 +15,8 @@
 import org.opensearch.sql.calcite.utils.binning.BinFieldValidator;
 import org.opensearch.sql.calcite.utils.binning.BinHandler;
 import org.opensearch.sql.calcite.utils.binning.BinnableField;
-import org.opensearch.sql.expression.function.PPLBuiltinOperators;
+import org.opensearch.sql.expression.function.BuiltinFunctionName;
+import org.opensearch.sql.expression.function.PPLFuncImpTable;
 
 /** Handler for bins-based (count) binning operations. */
 public class CountBinHandler implements BinHandler {
@@ -40,7 +40,9 @@ public RexNode createExpression(
     // Calculate data range using window functions
     RexNode minValue = context.relBuilder.min(fieldExpr).over().toRex();
     RexNode maxValue = context.relBuilder.max(fieldExpr).over().toRex();
-    RexNode dataRange = context.relBuilder.call(SqlStdOperatorTable.MINUS, maxValue, minValue);
+    RexNode dataRange =
+        PPLFuncImpTable.INSTANCE.resolve(
+            context.rexBuilder, BuiltinFunctionName.SUBTRACT, maxValue, minValue);
 
     // Convert start/end parameters
     RexNode startValue = convertParameter(countBin.getStart(), context);
@@ -49,8 +51,13 @@ public RexNode createExpression(
     // WIDTH_BUCKET(field_value, num_bins, data_range, max_value)
     RexNode numBins = context.relBuilder.literal(requestedBins);
 
-    return context.rexBuilder.makeCall(
-        PPLBuiltinOperators.WIDTH_BUCKET, fieldExpr, numBins, dataRange, maxValue);
+    return PPLFuncImpTable.INSTANCE.resolve(
+        context.rexBuilder,
+        BuiltinFunctionName.WIDTH_BUCKET,
+        fieldExpr,
+        numBins,
+        dataRange,
+        maxValue);
   }
 
   private RexNode convertParameter(

core/src/main/java/org/opensearch/sql/calcite/utils/binning/handlers/DefaultBinHandler.java

@@ -16,6 +16,8 @@
 import org.opensearch.sql.calcite.utils.binning.BinHandler;
 import org.opensearch.sql.calcite.utils.binning.BinnableField;
 import org.opensearch.sql.calcite.utils.binning.RangeFormatter;
+import org.opensearch.sql.expression.function.BuiltinFunctionName;
+import org.opensearch.sql.expression.function.PPLFuncImpTable;
 
 /** Handler for default binning when no parameters are specified. */
 public class DefaultBinHandler implements BinHandler {
@@ -45,7 +47,9 @@ private RexNode createNumericDefaultBinning(RexNode fieldExpr, CalcitePlanContex
     // Calculate data range
     RexNode minValue = context.relBuilder.min(fieldExpr).over().toRex();
     RexNode maxValue = context.relBuilder.max(fieldExpr).over().toRex();
-    RexNode dataRange = context.relBuilder.call(SqlStdOperatorTable.MINUS, maxValue, minValue);
+    RexNode dataRange =
+        PPLFuncImpTable.INSTANCE.resolve(
+            context.rexBuilder, BuiltinFunctionName.SUBTRACT, maxValue, minValue);
 
     // Calculate magnitude-based width
     RexNode log10Range = context.relBuilder.call(SqlStdOperatorTable.LOG10, dataRange);
@@ -60,17 +64,21 @@ private RexNode createNumericDefaultBinning(RexNode fieldExpr, CalcitePlanContex
     // Calculate bin value
     RexNode binStartValue = calculateBinValue(fieldExpr, widthInt, context);
     RexNode binEndValue =
-        context.relBuilder.call(SqlStdOperatorTable.PLUS, binStartValue, widthInt);
+        PPLFuncImpTable.INSTANCE.resolve(
+            context.rexBuilder, BuiltinFunctionName.ADD, binStartValue, widthInt);
 
     return RangeFormatter.createRangeString(binStartValue, binEndValue, context);
   }
 
   private RexNode calculateBinValue(RexNode fieldExpr, RexNode width, CalcitePlanContext context) {
 
-    RexNode divided = context.relBuilder.call(SqlStdOperatorTable.DIVIDE, fieldExpr, width);
+    RexNode divided =
+        PPLFuncImpTable.INSTANCE.resolve(
+            context.rexBuilder, BuiltinFunctionName.DIVIDE, fieldExpr, width);
 
     RexNode floored = context.relBuilder.call(SqlStdOperatorTable.FLOOR, divided);
 
-    return context.relBuilder.call(SqlStdOperatorTable.MULTIPLY, floored, width);
+    return PPLFuncImpTable.INSTANCE.resolve(
+        context.rexBuilder, BuiltinFunctionName.MULTIPLY, floored, width);
   }
 }

core/src/main/java/org/opensearch/sql/calcite/utils/binning/handlers/LogSpanHelper.java

@@ -11,6 +11,8 @@
 import org.opensearch.sql.calcite.utils.binning.BinConstants;
 import org.opensearch.sql.calcite.utils.binning.RangeFormatter;
 import org.opensearch.sql.calcite.utils.binning.SpanInfo;
+import org.opensearch.sql.expression.function.BuiltinFunctionName;
+import org.opensearch.sql.expression.function.PPLFuncImpTable;
 
 /** Helper for creating logarithmic span expressions. */
 public class LogSpanHelper {
@@ -31,14 +33,19 @@ public RexNode createLogSpanExpression(
     RexNode adjustedField = fieldExpr;
     if (coefficient != 1.0) {
       adjustedField =
-          context.relBuilder.call(
-              SqlStdOperatorTable.DIVIDE, fieldExpr, context.relBuilder.literal(coefficient));
+          PPLFuncImpTable.INSTANCE.resolve(
+              context.rexBuilder,
+              BuiltinFunctionName.DIVIDE,
+              fieldExpr,
+              context.relBuilder.literal(coefficient));
     }
 
     // Calculate log_base(adjusted_field)
     RexNode lnField = context.relBuilder.call(SqlStdOperatorTable.LN, adjustedField);
     RexNode lnBase = context.relBuilder.literal(Math.log(base));
-    RexNode logValue = context.relBuilder.call(SqlStdOperatorTable.DIVIDE, lnField, lnBase);
+    RexNode logValue =
+        PPLFuncImpTable.INSTANCE.resolve(
+            context.rexBuilder, BuiltinFunctionName.DIVIDE, lnField, lnBase);
 
     // Get bin number
     RexNode binNumber = context.relBuilder.call(SqlStdOperatorTable.FLOOR, logValue);
@@ -49,15 +56,20 @@ public RexNode createLogSpanExpression(
 
     RexNode basePowerBin = context.relBuilder.call(SqlStdOperatorTable.POWER, baseNode, binNumber);
     RexNode lowerBound =
-        context.relBuilder.call(SqlStdOperatorTable.MULTIPLY, coefficientNode, basePowerBin);
+        PPLFuncImpTable.INSTANCE.resolve(
+            context.rexBuilder, BuiltinFunctionName.MULTIPLY, coefficientNode, basePowerBin);
 
     RexNode binPlusOne =
-        context.relBuilder.call(
-            SqlStdOperatorTable.PLUS, binNumber, context.relBuilder.literal(1.0));
+        PPLFuncImpTable.INSTANCE.resolve(
+            context.rexBuilder,
+            BuiltinFunctionName.ADD,
+            binNumber,
+            context.relBuilder.literal(1.0));
     RexNode basePowerBinPlusOne =
         context.relBuilder.call(SqlStdOperatorTable.POWER, baseNode, binPlusOne);
     RexNode upperBound =
-        context.relBuilder.call(SqlStdOperatorTable.MULTIPLY, coefficientNode, basePowerBinPlusOne);
+        PPLFuncImpTable.INSTANCE.resolve(
+            context.rexBuilder, BuiltinFunctionName.MULTIPLY, coefficientNode, basePowerBinPlusOne);
 
     // Create range string
     RexNode rangeStr = RangeFormatter.createRangeString(lowerBound, upperBound, context);

core/src/main/java/org/opensearch/sql/calcite/utils/binning/handlers/MinSpanBinHandler.java

@@ -9,7 +9,6 @@
 
 import org.apache.calcite.rex.RexLiteral;
 import org.apache.calcite.rex.RexNode;
-import org.apache.calcite.sql.fun.SqlStdOperatorTable;
 import org.opensearch.sql.ast.expression.Literal;
 import org.opensearch.sql.ast.tree.Bin;
 import org.opensearch.sql.ast.tree.MinSpanBin;
@@ -18,7 +17,8 @@
 import org.opensearch.sql.calcite.utils.binning.BinFieldValidator;
 import org.opensearch.sql.calcite.utils.binning.BinHandler;
 import org.opensearch.sql.calcite.utils.binning.BinnableField;
-import org.opensearch.sql.expression.function.PPLBuiltinOperators;
+import org.opensearch.sql.expression.function.BuiltinFunctionName;
+import org.opensearch.sql.expression.function.PPLFuncImpTable;
 
 /** Handler for minspan-based binning operations. */
 public class MinSpanBinHandler implements BinHandler {
@@ -51,7 +51,9 @@ public RexNode createExpression(
     // Calculate data range using window functions
     RexNode minValue = context.relBuilder.min(fieldExpr).over().toRex();
     RexNode maxValue = context.relBuilder.max(fieldExpr).over().toRex();
-    RexNode dataRange = context.relBuilder.call(SqlStdOperatorTable.MINUS, maxValue, minValue);
+    RexNode dataRange =
+        PPLFuncImpTable.INSTANCE.resolve(
+            context.rexBuilder, BuiltinFunctionName.SUBTRACT, maxValue, minValue);
 
     // Convert start/end parameters
     RexNode startValue = convertParameter(minSpanBin.getStart(), context);
@@ -60,8 +62,13 @@ public RexNode createExpression(
     // MINSPAN_BUCKET(field_value, min_span, data_range, max_value)
     RexNode minSpanParam = context.relBuilder.literal(minspan);
 
-    return context.rexBuilder.makeCall(
-        PPLBuiltinOperators.MINSPAN_BUCKET, fieldExpr, minSpanParam, dataRange, maxValue);
+    return PPLFuncImpTable.INSTANCE.resolve(
+        context.rexBuilder,
+        BuiltinFunctionName.MINSPAN_BUCKET,
+        fieldExpr,
+        minSpanParam,
+        dataRange,
+        maxValue);
   }
 
   private RexNode convertParameter(

core/src/main/java/org/opensearch/sql/calcite/utils/binning/handlers/NumericSpanHelper.java

@@ -7,7 +7,8 @@
 
 import org.apache.calcite.rex.RexNode;
 import org.opensearch.sql.calcite.CalcitePlanContext;
-import org.opensearch.sql.expression.function.PPLBuiltinOperators;
+import org.opensearch.sql.expression.function.BuiltinFunctionName;
+import org.opensearch.sql.expression.function.PPLFuncImpTable;
 
 /** Helper for creating numeric span expressions. */
 public class NumericSpanHelper {
@@ -32,6 +33,7 @@ private RexNode createExpression(
       RexNode fieldExpr, RexNode spanValue, CalcitePlanContext context) {
 
     // SPAN_BUCKET(field_value, span_value)
-    return context.rexBuilder.makeCall(PPLBuiltinOperators.SPAN_BUCKET, fieldExpr, spanValue);
+    return PPLFuncImpTable.INSTANCE.resolve(
+        context.rexBuilder, BuiltinFunctionName.SPAN_BUCKET, fieldExpr, spanValue);
   }
 }

core/src/main/java/org/opensearch/sql/calcite/utils/binning/handlers/RangeBinHandler.java

@@ -13,7 +13,8 @@
 import org.opensearch.sql.calcite.utils.binning.BinFieldValidator;
 import org.opensearch.sql.calcite.utils.binning.BinHandler;
 import org.opensearch.sql.calcite.utils.binning.BinnableField;
-import org.opensearch.sql.expression.function.PPLBuiltinOperators;
+import org.opensearch.sql.expression.function.BuiltinFunctionName;
+import org.opensearch.sql.expression.function.PPLFuncImpTable;
 
 /** Handler for range-based binning (start/end parameters only). */
 public class RangeBinHandler implements BinHandler {
@@ -43,8 +44,14 @@ public RexNode createExpression(
     RexNode endParam = convertParameter(rangeBin.getEnd(), context, visitor);
 
     // Use RANGE_BUCKET with data bounds and user parameters
-    return context.rexBuilder.makeCall(
-        PPLBuiltinOperators.RANGE_BUCKET, fieldExpr, dataMin, dataMax, startParam, endParam);
+    return PPLFuncImpTable.INSTANCE.resolve(
+        context.rexBuilder,
+        BuiltinFunctionName.RANGE_BUCKET,
+        fieldExpr,
+        dataMin,
+        dataMax,
+        startParam,
+        endParam);
   }
 
   private RexNode convertParameter(