Fix string concat incompatibility from UTF-8 charset on literals

The previous fix added UTF-8 charset only to string literals in
visitLiteral(), leaving column VARCHAR types with no charset. Calcite
then rejected string concatenation (e.g. 'Hello ' + firstname) with:
VARCHAR CHARACTER SET "UTF-8" NOT NULL is not comparable to VARCHAR

Fix: move the UTF-8 + IMPLICIT collation enforcement into
OpenSearchTypeFactory.createSqlType() for VARCHAR/CHAR so both column
types and literal types carry the same charset consistently.
visitLiteral() reverts to plain createSqlType() calls since the factory
now handles encoding globally.

Signed-off-by: Radhakrishnan Pachyappan <gingeekrishna@gmail.com>

Author: gingeekrishna

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

@@ -12,7 +12,6 @@
 import static org.opensearch.sql.calcite.utils.OpenSearchTypeFactory.TYPE_FACTORY;
 
 import java.math.BigDecimal;
-import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
@@ -32,7 +31,6 @@
 import org.apache.calcite.rex.RexLambdaRef;
 import org.apache.calcite.rex.RexLiteral;
 import org.apache.calcite.rex.RexNode;
-import org.apache.calcite.sql.SqlCollation;
 import org.apache.calcite.sql.SqlIntervalQualifier;
 import org.apache.calcite.sql.SqlOperator;
 import org.apache.calcite.sql.fun.SqlStdOperatorTable;
@@ -133,28 +131,20 @@ public RexNode visitLiteral(Literal node, CalcitePlanContext context) {
       case NULL:
         return rexBuilder.makeNullLiteral(typeFactory.createSqlType(SqlTypeName.NULL));
       case STRING:
-        // Use UTF-8 explicitly so non-ASCII characters (e.g. Chinese, Arabic) are not rejected
-        // by Calcite's NlsString, which otherwise defaults to ISO-8859-1.
+        // UTF-8 charset is applied globally by OpenSearchTypeFactory.createSqlType() for
+        // VARCHAR/CHAR, so non-ASCII characters (e.g. Chinese, Arabic) are handled correctly
+        // and literal types stay compatible with column types for string operations.
         if (value.toString().length() == 1) {
           // To align Spark/PostgreSQL, Char(1) is useful, such as cast('1' to boolean) should
           // return true
           return rexBuilder.makeLiteral(
-              value.toString(),
-              typeFactory.createTypeWithCharsetAndCollation(
-                  typeFactory.createSqlType(SqlTypeName.CHAR),
-                  StandardCharsets.UTF_8,
-                  SqlCollation.IMPLICIT));
+              value.toString(), typeFactory.createSqlType(SqlTypeName.CHAR));
         } else {
           // Specific the type to VARCHAR and allowCast to true, or the STRING will be optimized to
           // CHAR(n)
           // which leads to incorrect return type in deriveReturnType of some functions/operators
           return rexBuilder.makeLiteral(
-              value.toString(),
-              typeFactory.createTypeWithCharsetAndCollation(
-                  typeFactory.createSqlType(SqlTypeName.VARCHAR),
-                  StandardCharsets.UTF_8,
-                  SqlCollation.IMPLICIT),
-              true);
+              value.toString(), typeFactory.createSqlType(SqlTypeName.VARCHAR), true);
         }
       case INTEGER:
         return rexBuilder.makeExactLiteral(new BigDecimal((Integer) value));

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

@@ -32,6 +32,7 @@
 
 import java.lang.reflect.Type;
 import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.LinkedHashMap;
 import java.util.List;
@@ -104,8 +105,21 @@ public RelDataType createTypeWithCharsetAndCollation(
     return super.createTypeWithCharsetAndCollation(type, charset, collation);
   }
 
+  @Override
+  public RelDataType createSqlType(SqlTypeName typeName) {
+    RelDataType type = super.createSqlType(typeName);
+    if (typeName == SqlTypeName.VARCHAR || typeName == SqlTypeName.CHAR) {
+      return createTypeWithCharsetAndCollation(type, StandardCharsets.UTF_8, SqlCollation.IMPLICIT);
+    }
+    return type;
+  }
+
   public RelDataType createSqlType(SqlTypeName typeName, boolean nullable) {
-    return createTypeWithNullability(super.createSqlType(typeName), nullable);
+    RelDataType type = createTypeWithNullability(super.createSqlType(typeName), nullable);
+    if (typeName == SqlTypeName.VARCHAR || typeName == SqlTypeName.CHAR) {
+      return createTypeWithCharsetAndCollation(type, StandardCharsets.UTF_8, SqlCollation.IMPLICIT);
+    }
+    return type;
   }
 
   public RelDataType createStructType(