From d1cc17c5792b6284b29f7ed05a3dea01ac27ca29 Mon Sep 17 00:00:00 2001
From: Rong Rong <rongr@apache.org>
Date: Wed, 3 Jan 2024 13:13:41 -0800
Subject: [PATCH] [multistage][hotfix] use UTF-8 as default CharSet, this is
 also true for v1 engine (#12213)

* [hotfix] use UTF-8 as default CharSet, this is also true for v1 engine

---------

Co-authored-by: Rong Rong <rongr@startree.ai>
---
 .../apache/pinot/query/type/TypeFactory.java  |   8 ++
 .../pinot/query/type/TypeFactoryTest.java     |  46 ++++--
 .../resources/queries/AggregatePlans.json     |  14 +-
 .../resources/queries/BasicQueryPlans.json    |  10 +-
 .../test/resources/queries/GroupByPlans.json  |  24 ++--
 .../src/test/resources/queries/JoinPlans.json |  48 +++----
 .../queries/LiteralEvaluationPlans.json       |  10 +-
 .../resources/queries/PinotHintablePlans.json |  10 +-
 .../queries/WindowFunctionPlans.json          | 134 +++++++++---------
 .../resources/queries/CharacterTypes.json     |   2 -
 10 files changed, 164 insertions(+), 142 deletions(-)

diff --git a/pinot-query-planner/src/main/java/org/apache/pinot/query/type/TypeFactory.java b/pinot-query-planner/src/main/java/org/apache/pinot/query/type/TypeFactory.java
index ccc0f41a080f..054d19113492 100644
--- a/pinot-query-planner/src/main/java/org/apache/pinot/query/type/TypeFactory.java
+++ b/pinot-query-planner/src/main/java/org/apache/pinot/query/type/TypeFactory.java
@@ -18,6 +18,8 @@
  */
 package org.apache.pinot.query.type;
 
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
 import java.util.Map;
 import java.util.function.Predicate;
 import org.apache.calcite.jdbc.JavaTypeFactoryImpl;
@@ -39,11 +41,17 @@
  * upgrading Calcite versions.
  */
 public class TypeFactory extends JavaTypeFactoryImpl {
+  private static final Charset DEFAULT_CHARSET = StandardCharsets.UTF_8;
 
   public TypeFactory(RelDataTypeSystem typeSystem) {
     super(typeSystem);
   }
 
+  @Override
+  public Charset getDefaultCharset() {
+    return DEFAULT_CHARSET;
+  }
+
   public RelDataType createRelDataTypeFromSchema(Schema schema) {
     Builder builder = new Builder(this);
     Predicate<FieldSpec> isNullable;
diff --git a/pinot-query-planner/src/test/java/org/apache/pinot/query/type/TypeFactoryTest.java b/pinot-query-planner/src/test/java/org/apache/pinot/query/type/TypeFactoryTest.java
index 8c1686b2c8c6..543e1c3b70e7 100644
--- a/pinot-query-planner/src/test/java/org/apache/pinot/query/type/TypeFactoryTest.java
+++ b/pinot-query-planner/src/test/java/org/apache/pinot/query/type/TypeFactoryTest.java
@@ -18,6 +18,8 @@
  */
 package org.apache.pinot.query.type;
 
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
@@ -25,6 +27,7 @@
 import org.apache.calcite.jdbc.JavaTypeFactoryImpl;
 import org.apache.calcite.rel.type.RelDataType;
 import org.apache.calcite.rel.type.RelDataTypeField;
+import org.apache.calcite.sql.SqlCollation;
 import org.apache.calcite.sql.type.ArraySqlType;
 import org.apache.calcite.sql.type.BasicSqlType;
 import org.apache.calcite.sql.type.SqlTypeName;
@@ -37,23 +40,22 @@
 
 public class TypeFactoryTest {
   private static final TypeSystem TYPE_SYSTEM = new TypeSystem();
+  private static final JavaTypeFactory TYPE_FACTORY = new TestJavaTypeFactoryImpl(TYPE_SYSTEM);
 
   @DataProvider(name = "relDataTypeConversion")
   public Iterator<Object[]> relDataTypeConversion() {
     ArrayList<Object[]> cases = new ArrayList<>();
 
-    JavaTypeFactory javaTypeFactory = new JavaTypeFactoryImpl(TYPE_SYSTEM);
-
     for (FieldSpec.DataType dataType : FieldSpec.DataType.values()) {
       RelDataType basicType;
       RelDataType arrayType = null;
       switch (dataType) {
         case INT: {
-          basicType = javaTypeFactory.createSqlType(SqlTypeName.INTEGER);
+          basicType = TYPE_FACTORY.createSqlType(SqlTypeName.INTEGER);
           break;
         }
         case LONG: {
-          basicType = javaTypeFactory.createSqlType(SqlTypeName.BIGINT);
+          basicType = TYPE_FACTORY.createSqlType(SqlTypeName.BIGINT);
           break;
         }
         // Map float and double to the same RelDataType so that queries like
@@ -68,33 +70,33 @@ public Iterator<Object[]> relDataTypeConversion() {
         // With float and double mapped to the same RelDataType, the behavior in multi-stage query engine will be the
         // same as the query in v1 query engine.
         case FLOAT: {
-          basicType = javaTypeFactory.createSqlType(SqlTypeName.DOUBLE);
-          arrayType = javaTypeFactory.createSqlType(SqlTypeName.REAL);
+          basicType = TYPE_FACTORY.createSqlType(SqlTypeName.DOUBLE);
+          arrayType = TYPE_FACTORY.createSqlType(SqlTypeName.REAL);
           break;
         }
         case DOUBLE: {
-          basicType = javaTypeFactory.createSqlType(SqlTypeName.DOUBLE);
+          basicType = TYPE_FACTORY.createSqlType(SqlTypeName.DOUBLE);
           break;
         }
         case BOOLEAN: {
-          basicType = javaTypeFactory.createSqlType(SqlTypeName.BOOLEAN);
+          basicType = TYPE_FACTORY.createSqlType(SqlTypeName.BOOLEAN);
           break;
         }
         case TIMESTAMP: {
-          basicType = javaTypeFactory.createSqlType(SqlTypeName.TIMESTAMP);
+          basicType = TYPE_FACTORY.createSqlType(SqlTypeName.TIMESTAMP);
           break;
         }
         case STRING:
         case JSON: {
-          basicType = javaTypeFactory.createSqlType(SqlTypeName.VARCHAR);
+          basicType = TYPE_FACTORY.createSqlType(SqlTypeName.VARCHAR);
           break;
         }
         case BYTES: {
-          basicType = javaTypeFactory.createSqlType(SqlTypeName.VARBINARY);
+          basicType = TYPE_FACTORY.createSqlType(SqlTypeName.VARBINARY);
           break;
         }
         case BIG_DECIMAL: {
-          basicType = javaTypeFactory.createSqlType(SqlTypeName.DECIMAL);
+          basicType = TYPE_FACTORY.createSqlType(SqlTypeName.DECIMAL);
           break;
         }
         case LIST:
@@ -268,7 +270,9 @@ public void testRelDataTypeConversion() {
           break;
         case "STRING_COL":
         case "JSON_COL":
-          Assert.assertEquals(field.getType(), new BasicSqlType(TYPE_SYSTEM, SqlTypeName.VARCHAR));
+          Assert.assertEquals(field.getType(),
+              TYPE_FACTORY.createTypeWithCharsetAndCollation(new BasicSqlType(TYPE_SYSTEM, SqlTypeName.VARCHAR),
+                  StandardCharsets.UTF_8, SqlCollation.IMPLICIT));
           break;
         case "BYTES_COL":
           Assert.assertEquals(field.getType(), new BasicSqlType(TYPE_SYSTEM, SqlTypeName.VARBINARY));
@@ -290,8 +294,9 @@ public void testRelDataTypeConversion() {
               new ArraySqlType(new BasicSqlType(TYPE_SYSTEM, SqlTypeName.DOUBLE), false));
           break;
         case "STRING_ARRAY_COL":
-          Assert.assertEquals(field.getType(),
-              new ArraySqlType(new BasicSqlType(TYPE_SYSTEM, SqlTypeName.VARCHAR), false));
+          Assert.assertEquals(field.getType(), new ArraySqlType(
+              TYPE_FACTORY.createTypeWithCharsetAndCollation(new BasicSqlType(TYPE_SYSTEM, SqlTypeName.VARCHAR),
+                  StandardCharsets.UTF_8, SqlCollation.IMPLICIT), false));
           break;
         case "BYTES_ARRAY_COL":
           Assert.assertEquals(field.getType(),
@@ -304,6 +309,17 @@ public void testRelDataTypeConversion() {
     }
   }
 
+  private static class TestJavaTypeFactoryImpl extends JavaTypeFactoryImpl {
+    public TestJavaTypeFactoryImpl(TypeSystem typeSystem) {
+      super(typeSystem);
+    }
+
+    @Override
+    public Charset getDefaultCharset() {
+      return StandardCharsets.UTF_8;
+    }
+  }
+
   //tests precision and scale for numeric data type
   private void checkPrecisionScale(RelDataTypeField field, BasicSqlType basicSqlType) {
     Assert.assertEquals(field.getValue().getPrecision(), basicSqlType.getPrecision());
diff --git a/pinot-query-planner/src/test/resources/queries/AggregatePlans.json b/pinot-query-planner/src/test/resources/queries/AggregatePlans.json
index 9a8f0424069b..1072ccd03c84 100644
--- a/pinot-query-planner/src/test/resources/queries/AggregatePlans.json
+++ b/pinot-query-planner/src/test/resources/queries/AggregatePlans.json
@@ -32,7 +32,7 @@
           "\n  LogicalAggregate(group=[{}], agg#0=[$SUM0($0)], agg#1=[COUNT($1)])",
           "\n    PinotLogicalExchange(distribution=[hash])",
           "\n      LogicalAggregate(group=[{}], agg#0=[$SUM0($3)], agg#1=[COUNT()])",
-          "\n        LogicalFilter(condition=[AND(>=($2, 0), =($1, 'pink floyd'))])",
+          "\n        LogicalFilter(condition=[AND(>=($2, 0), =($1, _UTF-8'pink floyd'))])",
           "\n          LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -46,7 +46,7 @@
           "\n  LogicalAggregate(group=[{}], agg#0=[$SUM0($0)], agg#1=[COUNT($1)], agg#2=[MAX($2)])",
           "\n    PinotLogicalExchange(distribution=[hash])",
           "\n      LogicalAggregate(group=[{}], agg#0=[$SUM0($3)], agg#1=[COUNT()], agg#2=[MAX($3)])",
-          "\n        LogicalFilter(condition=[AND(>=($2, 0), =($1, 'pink floyd'))])",
+          "\n        LogicalFilter(condition=[AND(>=($2, 0), =($1, _UTF-8'pink floyd'))])",
           "\n          LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -60,7 +60,7 @@
           "\n  LogicalAggregate(group=[{}], agg#0=[$SUM0($0)], agg#1=[COUNT($1)])",
           "\n    PinotLogicalExchange(distribution=[hash])",
           "\n      LogicalAggregate(group=[{}], agg#0=[$SUM0($2)], agg#1=[COUNT()])",
-          "\n        LogicalFilter(condition=[AND(>=($2, 0), =($1, 'pink floyd'))])",
+          "\n        LogicalFilter(condition=[AND(>=($2, 0), =($1, _UTF-8'pink floyd'))])",
           "\n          LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -87,7 +87,7 @@
           "\n  LogicalAggregate(group=[{}], agg#0=[$SUM0($0)], agg#1=[COUNT($1)])",
           "\n    PinotLogicalExchange(distribution=[hash])",
           "\n      LogicalAggregate(group=[{}], agg#0=[$SUM0($2)], agg#1=[COUNT()])",
-          "\n        LogicalFilter(condition=[AND(>=($2, 0), =($1, 'a'))])",
+          "\n        LogicalFilter(condition=[AND(>=($2, 0), =($1, _UTF-8'a'))])",
           "\n          LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -101,7 +101,7 @@
           "\n  LogicalAggregate(group=[{}], agg#0=[$SUM0($0)], agg#1=[COUNT($1)])",
           "\n    PinotLogicalExchange(distribution=[hash])",
           "\n      LogicalAggregate(group=[{}], agg#0=[$SUM0($2)], agg#1=[COUNT()])",
-          "\n        LogicalFilter(condition=[AND(>=($2, 0), =($1, 'pink floyd'))])",
+          "\n        LogicalFilter(condition=[AND(>=($2, 0), =($1, _UTF-8'pink floyd'))])",
           "\n          LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -115,7 +115,7 @@
           "\n  LogicalAggregate(group=[{}], agg#0=[$SUM0($0)], agg#1=[COUNT($1)])",
           "\n    PinotLogicalExchange(distribution=[hash])",
           "\n      LogicalAggregate(group=[{}], agg#0=[$SUM0($2)], agg#1=[COUNT()])",
-          "\n        LogicalFilter(condition=[AND(>=($2, 0), =($1, 'pink floyd'))])",
+          "\n        LogicalFilter(condition=[AND(>=($2, 0), =($1, _UTF-8'pink floyd'))])",
           "\n          LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -129,7 +129,7 @@
           "\n  LogicalAggregate(group=[{}], agg#0=[$SUM0($0)], agg#1=[COUNT($1)])",
           "\n    PinotLogicalExchange(distribution=[hash])",
           "\n      LogicalAggregate(group=[{}], agg#0=[$SUM0($2)], agg#1=[COUNT()])",
-          "\n        LogicalFilter(condition=[AND(>=($2, 0), =($1, 'pink floyd'))])",
+          "\n        LogicalFilter(condition=[AND(>=($2, 0), =($1, _UTF-8'pink floyd'))])",
           "\n          LogicalTableScan(table=[[a]])",
           "\n"
         ]
diff --git a/pinot-query-planner/src/test/resources/queries/BasicQueryPlans.json b/pinot-query-planner/src/test/resources/queries/BasicQueryPlans.json
index 8cbd90fde57b..59acc30e6611 100644
--- a/pinot-query-planner/src/test/resources/queries/BasicQueryPlans.json
+++ b/pinot-query-planner/src/test/resources/queries/BasicQueryPlans.json
@@ -40,7 +40,7 @@
         "output": [
           "Execution Plan",
           "\nLogicalProject(col1=[$0], EXPR$1=[+($2, $6)])",
-          "\n  LogicalFilter(condition=[AND(>=($2, 0), =($1, 'a'))])",
+          "\n  LogicalFilter(condition=[AND(>=($2, 0), =($1, _UTF-8'a'))])",
           "\n    LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -51,7 +51,7 @@
         "output": [
           "Execution Plan",
           "\nLogicalProject(col1=[$0], colsum=[+($2, $6)])",
-          "\n  LogicalFilter(condition=[AND(>=($2, 0), =($1, 'a'))])",
+          "\n  LogicalFilter(condition=[AND(>=($2, 0), =($1, _UTF-8'a'))])",
           "\n    LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -64,7 +64,7 @@
           "\nLogicalSort(offset=[0], fetch=[10])",
           "\n  PinotLogicalSortExchange(distribution=[hash], collation=[[]], isSortOnSender=[false], isSortOnReceiver=[false])",
           "\n    LogicalSort(fetch=[10])",
-          "\n      LogicalProject(EXPR$0=[DATETRUNC('DAY', $6)])",
+          "\n      LogicalProject(EXPR$0=[DATETRUNC(_UTF-8'DAY', $6)])",
           "\n        LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -77,7 +77,7 @@
           "\nLogicalSort(offset=[0], fetch=[10])",
           "\n  PinotLogicalSortExchange(distribution=[hash], collation=[[]], isSortOnSender=[false], isSortOnReceiver=[false])",
           "\n    LogicalSort(fetch=[10])",
-          "\n      LogicalProject(day=[DATETRUNC('DAY', $6)])",
+          "\n      LogicalProject(day=[DATETRUNC(_UTF-8'DAY', $6)])",
           "\n        LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -91,7 +91,7 @@
           "\n  LogicalAggregate(group=[{}], agg#0=[$SUM0($0)], agg#1=[COUNT($1)])",
           "\n    PinotLogicalExchange(distribution=[hash])",
           "\n      LogicalAggregate(group=[{}], agg#0=[$SUM0($0)], agg#1=[COUNT()])",
-          "\n        LogicalProject($f0=[CAST(CASE(>($2, 10), '1':VARCHAR, >($2, 20), '2':VARCHAR, >($2, 30), '3':VARCHAR, >($2, 40), '4':VARCHAR, >($2, 50), '5':VARCHAR, '0':VARCHAR)):DECIMAL(1000, 500) NOT NULL])",
+          "\n        LogicalProject($f0=[CAST(CASE(>($2, 10), _UTF-8'1':VARCHAR CHARACTER SET \"UTF-8\", >($2, 20), _UTF-8'2':VARCHAR CHARACTER SET \"UTF-8\", >($2, 30), _UTF-8'3':VARCHAR CHARACTER SET \"UTF-8\", >($2, 40), _UTF-8'4':VARCHAR CHARACTER SET \"UTF-8\", >($2, 50), _UTF-8'5':VARCHAR CHARACTER SET \"UTF-8\", _UTF-8'0':VARCHAR CHARACTER SET \"UTF-8\")):DECIMAL(1000, 500) NOT NULL])",
           "\n          LogicalTableScan(table=[[a]])",
           "\n"
         ]
diff --git a/pinot-query-planner/src/test/resources/queries/GroupByPlans.json b/pinot-query-planner/src/test/resources/queries/GroupByPlans.json
index 690acbd01f85..2a67b4455ac7 100644
--- a/pinot-query-planner/src/test/resources/queries/GroupByPlans.json
+++ b/pinot-query-planner/src/test/resources/queries/GroupByPlans.json
@@ -34,7 +34,7 @@
           "\nLogicalAggregate(group=[{0}], agg#0=[$SUM0($1)])",
           "\n  PinotLogicalExchange(distribution=[hash[0]])",
           "\n    LogicalAggregate(group=[{0}], agg#0=[$SUM0($2)])",
-          "\n      LogicalFilter(condition=[AND(>=($2, 0), =($1, 'a'))])",
+          "\n      LogicalFilter(condition=[AND(>=($2, 0), =($1, _UTF-8'a'))])",
           "\n        LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -48,7 +48,7 @@
           "\nLogicalAggregate(group=[{0}], agg#0=[COUNT($1)])",
           "\n  PinotLogicalExchange(distribution=[hash[0]])",
           "\n    LogicalAggregate(group=[{0}], agg#0=[COUNT()])",
-          "\n      LogicalFilter(condition=[AND(>=($2, 0), =($1, 'a'))])",
+          "\n      LogicalFilter(condition=[AND(>=($2, 0), =($1, _UTF-8'a'))])",
           "\n        LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -62,7 +62,7 @@
           "\n  LogicalAggregate(group=[{0, 1}], agg#0=[$SUM0($2)])",
           "\n    PinotLogicalExchange(distribution=[hash[0, 1]])",
           "\n      LogicalAggregate(group=[{0, 1}], agg#0=[$SUM0($2)])",
-          "\n        LogicalFilter(condition=[AND(>=($2, 0), =($0, 'a'))])",
+          "\n        LogicalFilter(condition=[AND(>=($2, 0), =($0, _UTF-8'a'))])",
           "\n          LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -77,7 +77,7 @@
           "\n    LogicalAggregate(group=[{0}], agg#0=[COUNT($1)], agg#1=[$SUM0($2)], agg#2=[MAX($3)], agg#3=[MIN($4)])",
           "\n      PinotLogicalExchange(distribution=[hash[0]])",
           "\n        LogicalAggregate(group=[{0}], agg#0=[COUNT()], agg#1=[$SUM0($2)], agg#2=[MAX($2)], agg#3=[MIN($2)])",
-          "\n          LogicalFilter(condition=[AND(>=($2, 0), =($1, 'a'))])",
+          "\n          LogicalFilter(condition=[AND(>=($2, 0), =($1, _UTF-8'a'))])",
           "\n            LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -92,7 +92,7 @@
           "\n    LogicalAggregate(group=[{0}], agg#0=[COUNT($1)], agg#1=[$SUM0($2)], agg#2=[MAX($3)], agg#3=[MIN($4)])",
           "\n      PinotLogicalExchange(distribution=[hash[0]])",
           "\n        LogicalAggregate(group=[{0}], agg#0=[COUNT()], agg#1=[$SUM0($2)], agg#2=[MAX($2)], agg#3=[MIN($2)])",
-          "\n          LogicalFilter(condition=[AND(>=($2, 0), =($1, 'a'))])",
+          "\n          LogicalFilter(condition=[AND(>=($2, 0), =($1, _UTF-8'a'))])",
           "\n            LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -143,7 +143,7 @@
           "\nLogicalAggregate(group=[{0}], EXPR$1=[$SUM0($1)])",
           "\n  PinotLogicalExchange(distribution=[hash[0]])",
           "\n    LogicalProject(col1=[$0], col3=[$2])",
-          "\n      LogicalFilter(condition=[AND(>=($2, 0), =($1, 'a'))])",
+          "\n      LogicalFilter(condition=[AND(>=($2, 0), =($1, _UTF-8'a'))])",
           "\n        LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -156,7 +156,7 @@
           "\nLogicalAggregate(group=[{0}], EXPR$1=[$SUM0($1)], EXPR$2=[MAX($1)])",
           "\n  PinotLogicalExchange(distribution=[hash[0]])",
           "\n    LogicalProject(col1=[$0], col3=[$2])",
-          "\n      LogicalFilter(condition=[AND(>=($2, 0), =($1, 'a'))])",
+          "\n      LogicalFilter(condition=[AND(>=($2, 0), =($1, _UTF-8'a'))])",
           "\n        LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -170,7 +170,7 @@
           "\nLogicalAggregate(group=[{0}], EXPR$1=[COUNT()])",
           "\n  PinotLogicalExchange(distribution=[hash[0]])",
           "\n    LogicalProject(col1=[$0])",
-          "\n      LogicalFilter(condition=[AND(>=($2, 0), =($1, 'a'))])",
+          "\n      LogicalFilter(condition=[AND(>=($2, 0), =($1, _UTF-8'a'))])",
           "\n        LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -184,7 +184,7 @@
           "\n  LogicalAggregate(group=[{0, 1}], EXPR$2=[$SUM0($2)])",
           "\n    PinotLogicalExchange(distribution=[hash[0, 1]])",
           "\n      LogicalProject(col1=[$0], col2=[$1], col3=[$2])",
-          "\n        LogicalFilter(condition=[AND(>=($2, 0), =($0, 'a'))])",
+          "\n        LogicalFilter(condition=[AND(>=($2, 0), =($0, _UTF-8'a'))])",
           "\n          LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -199,7 +199,7 @@
           "\n    LogicalAggregate(group=[{0}], EXPR$1=[COUNT()], EXPR$2=[$SUM0($1)], agg#2=[MAX($1)], agg#3=[MIN($1)])",
           "\n      PinotLogicalExchange(distribution=[hash[0]])",
           "\n        LogicalProject(col1=[$0], col3=[$2])",
-          "\n          LogicalFilter(condition=[AND(>=($2, 0), =($1, 'a'))])",
+          "\n          LogicalFilter(condition=[AND(>=($2, 0), =($1, _UTF-8'a'))])",
           "\n            LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -214,7 +214,7 @@
           "\n    LogicalAggregate(group=[{0}], EXPR$1=[$SUM0($1)], agg#1=[MAX($1)], agg#2=[MIN($1)], agg#3=[COUNT()])",
           "\n      PinotLogicalExchange(distribution=[hash[0]])",
           "\n        LogicalProject(col1=[$0], col3=[$2])",
-          "\n          LogicalFilter(condition=[AND(>=($2, 0), =($1, 'a'))])",
+          "\n          LogicalFilter(condition=[AND(>=($2, 0), =($1, _UTF-8'a'))])",
           "\n            LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -229,7 +229,7 @@
           "\n    LogicalAggregate(group=[{0}], count=[COUNT()], SUM=[$SUM0($1)], agg#2=[MAX($1)], agg#3=[MIN($1)])",
           "\n      PinotLogicalExchange(distribution=[hash[0]])",
           "\n        LogicalProject(col1=[$0], col3=[$2])",
-          "\n          LogicalFilter(condition=[AND(>=($2, 0), =($1, 'a'))])",
+          "\n          LogicalFilter(condition=[AND(>=($2, 0), =($1, _UTF-8'a'))])",
           "\n            LogicalTableScan(table=[[a]])",
           "\n"
         ]
diff --git a/pinot-query-planner/src/test/resources/queries/JoinPlans.json b/pinot-query-planner/src/test/resources/queries/JoinPlans.json
index 41ba0c9618fd..d9eb08b2ec71 100644
--- a/pinot-query-planner/src/test/resources/queries/JoinPlans.json
+++ b/pinot-query-planner/src/test/resources/queries/JoinPlans.json
@@ -102,7 +102,7 @@
           "\n  LogicalJoin(condition=[=($0, $2)], joinType=[inner])",
           "\n    PinotLogicalExchange(distribution=[hash[0]])",
           "\n      LogicalProject(col1=[$0], ts=[$6])",
-          "\n        LogicalFilter(condition=[AND(>=($2, 0), =($1, 'a'))])",
+          "\n        LogicalFilter(condition=[AND(>=($2, 0), =($1, _UTF-8'a'))])",
           "\n          LogicalTableScan(table=[[a]])",
           "\n    PinotLogicalExchange(distribution=[hash[0]])",
           "\n      LogicalProject(col2=[$1], col3=[$2])",
@@ -123,7 +123,7 @@
           "\n        LogicalJoin(condition=[=($0, $1)], joinType=[inner])",
           "\n          PinotLogicalExchange(distribution=[hash[0]])",
           "\n            LogicalProject(col1=[$0])",
-          "\n              LogicalFilter(condition=[AND(>=($2, 0), =($1, 'a'))])",
+          "\n              LogicalFilter(condition=[AND(>=($2, 0), =($1, _UTF-8'a'))])",
           "\n                LogicalTableScan(table=[[a]])",
           "\n          PinotLogicalExchange(distribution=[hash[0]])",
           "\n            LogicalProject(col2=[$1], col3=[$2])",
@@ -137,7 +137,7 @@
         "sql": "EXPLAIN PLAN FOR SELECT dateTrunc('DAY', a.ts + b.ts) FROM a JOIN b on a.col1 = b.col1 AND a.col2 = b.col2",
         "output": [
           "Execution Plan",
-          "\nLogicalProject(EXPR$0=[DATETRUNC('DAY', +($2, $5))])",
+          "\nLogicalProject(EXPR$0=[DATETRUNC(_UTF-8'DAY', +($2, $5))])",
           "\n  LogicalJoin(condition=[AND(=($0, $3), =($1, $4))], joinType=[inner])",
           "\n    PinotLogicalExchange(distribution=[hash[0, 1]])",
           "\n      LogicalProject(col1=[$0], col2=[$1], ts=[$6])",
@@ -195,11 +195,11 @@
           "\n  LogicalJoin(condition=[=($0, $1)], joinType=[inner])",
           "\n    PinotLogicalExchange(distribution=[hash[0]])",
           "\n      LogicalProject(col1=[$0])",
-          "\n        LogicalFilter(condition=[OR(=($1, 'bar'), =($1, 'foo'))])",
+          "\n        LogicalFilter(condition=[OR(=($1, _UTF-8'bar'), =($1, _UTF-8'foo'))])",
           "\n          LogicalTableScan(table=[[a]])",
           "\n    PinotLogicalExchange(distribution=[hash[0]])",
           "\n      LogicalProject(col1=[$0], col2=[$1])",
-          "\n        LogicalFilter(condition=[AND(<>($1, 'alice':VARCHAR(7)), <>($1, 'charlie':VARCHAR(7)))])",
+          "\n        LogicalFilter(condition=[AND(<>($1, _UTF-8'alice':VARCHAR(7) CHARACTER SET \"UTF-8\"), <>($1, _UTF-8'charlie':VARCHAR(7) CHARACTER SET \"UTF-8\"))])",
           "\n          LogicalTableScan(table=[[b]])",
           "\n"
         ]
@@ -213,11 +213,11 @@
           "\n  LogicalJoin(condition=[=($0, $1)], joinType=[inner])",
           "\n    PinotLogicalExchange(distribution=[hash[0]])",
           "\n      LogicalProject(col1=[$0])",
-          "\n        LogicalFilter(condition=[OR(=($1, 'bar'), =($1, 'foo'))])",
+          "\n        LogicalFilter(condition=[OR(=($1, _UTF-8'bar'), =($1, _UTF-8'foo'))])",
           "\n          LogicalTableScan(table=[[a]])",
           "\n    PinotLogicalExchange(distribution=[hash[0]])",
           "\n      LogicalProject(col1=[$0], col2=[$1])",
-          "\n        LogicalFilter(condition=[AND(<>($1, 'alice':VARCHAR(7)), <>($1, 'charlie':VARCHAR(7)))])",
+          "\n        LogicalFilter(condition=[AND(<>($1, _UTF-8'alice':VARCHAR(7) CHARACTER SET \"UTF-8\"), <>($1, _UTF-8'charlie':VARCHAR(7) CHARACTER SET \"UTF-8\"))])",
           "\n          LogicalTableScan(table=[[b]])",
           "\n"
         ]
@@ -251,19 +251,19 @@
           "\n          LogicalJoin(condition=[=($2, $3)], joinType=[semi])",
           "\n            PinotLogicalExchange(distribution=[hash[2]])",
           "\n              LogicalProject(col1=[$0], col2=[$1], col3=[$2])",
-          "\n                LogicalFilter(condition=[=($1, 'test')])",
+          "\n                LogicalFilter(condition=[=($1, _UTF-8'test')])",
           "\n                  LogicalTableScan(table=[[a]])",
           "\n            PinotLogicalExchange(distribution=[hash[0]])",
           "\n              LogicalProject(col3=[$2])",
-          "\n                LogicalFilter(condition=[=($0, 'foo')])",
+          "\n                LogicalFilter(condition=[=($0, _UTF-8'foo')])",
           "\n                  LogicalTableScan(table=[[b]])",
           "\n        PinotLogicalExchange(distribution=[hash[0]])",
           "\n          LogicalProject(col3=[$2])",
-          "\n            LogicalFilter(condition=[=($0, 'bar')])",
+          "\n            LogicalFilter(condition=[=($0, _UTF-8'bar')])",
           "\n              LogicalTableScan(table=[[b]])",
           "\n    PinotLogicalExchange(distribution=[hash[0]])",
           "\n      LogicalProject(col3=[$2])",
-          "\n        LogicalFilter(condition=[=($0, 'foobar')])",
+          "\n        LogicalFilter(condition=[=($0, _UTF-8'foobar')])",
           "\n          LogicalTableScan(table=[[b]])",
           "\n"
         ]
@@ -286,28 +286,28 @@
           "\n                    LogicalJoin(condition=[=($3, $4)], joinType=[left])",
           "\n                      PinotLogicalExchange(distribution=[hash[3]])",
           "\n                        LogicalProject(col1=[$0], col2=[$1], col3=[$2], col30=[$2])",
-          "\n                          LogicalFilter(condition=[=($1, 'test')])",
+          "\n                          LogicalFilter(condition=[=($1, _UTF-8'test')])",
           "\n                            LogicalTableScan(table=[[a]])",
           "\n                      PinotLogicalExchange(distribution=[hash[0]])",
           "\n                        LogicalAggregate(group=[{0}], agg#0=[MIN($1)])",
           "\n                          PinotLogicalExchange(distribution=[hash[0]])",
           "\n                            LogicalAggregate(group=[{0}], agg#0=[MIN($1)])",
           "\n                              LogicalProject(col3=[$2], $f1=[true])",
-          "\n                                LogicalFilter(condition=[=($0, 'foo')])",
+          "\n                                LogicalFilter(condition=[=($0, _UTF-8'foo')])",
           "\n                                  LogicalTableScan(table=[[b]])",
           "\n              PinotLogicalExchange(distribution=[hash[0]])",
           "\n                LogicalAggregate(group=[{0}], agg#0=[MIN($1)])",
           "\n                  PinotLogicalExchange(distribution=[hash[0]])",
           "\n                    LogicalAggregate(group=[{0}], agg#0=[MIN($1)])",
           "\n                      LogicalProject(col3=[$2], $f1=[true])",
-          "\n                        LogicalFilter(condition=[=($0, 'bar')])",
+          "\n                        LogicalFilter(condition=[=($0, _UTF-8'bar')])",
           "\n                          LogicalTableScan(table=[[b]])",
           "\n      PinotLogicalExchange(distribution=[hash[0]])",
           "\n        LogicalAggregate(group=[{0}], agg#0=[MIN($1)])",
           "\n          PinotLogicalExchange(distribution=[hash[0]])",
           "\n            LogicalAggregate(group=[{0}], agg#0=[MIN($1)])",
           "\n              LogicalProject(col3=[$2], $f1=[true])",
-          "\n                LogicalFilter(condition=[=($0, 'foobar')])",
+          "\n                LogicalFilter(condition=[=($0, _UTF-8'foobar')])",
           "\n                  LogicalTableScan(table=[[b]])",
           "\n"
         ]
@@ -322,7 +322,7 @@
           "\n    PinotLogicalExchange(distribution=[hash[0, 1]])",
           "\n      LogicalTableScan(table=[[a]])",
           "\n    PinotLogicalExchange(distribution=[hash[0, 1]])",
-          "\n      LogicalProject(col1=[$0], col2=[$1], col10=[CAST($0):VARCHAR], col20=[CAST($1):VARCHAR], $f2=[CAST($2):BIGINT], EXPR$3=[*(0.5:DECIMAL(2, 1), $2)])",
+          "\n      LogicalProject(col1=[$0], col2=[$1], col10=[CAST($0):VARCHAR CHARACTER SET \"UTF-8\"], col20=[CAST($1):VARCHAR CHARACTER SET \"UTF-8\"], $f2=[CAST($2):BIGINT], EXPR$3=[*(0.5:DECIMAL(2, 1), $2)])",
           "\n        LogicalAggregate(group=[{0, 1}], agg#0=[$SUM0($2)])",
           "\n          PinotLogicalExchange(distribution=[hash[0, 1]])",
           "\n            LogicalAggregate(group=[{0, 1}], agg#0=[$SUM0($2)])",
@@ -340,7 +340,7 @@
           "\n    PinotLogicalExchange(distribution=[hash[0, 1]])",
           "\n      LogicalTableScan(table=[[a]])",
           "\n    PinotLogicalExchange(distribution=[hash[0, 1]])",
-          "\n      LogicalProject(col1=[$0], col2=[$1], EXPR$0=[$2], col10=[CAST($0):VARCHAR], col20=[CAST($1):VARCHAR], EXPR$05=[CAST($2):DECIMAL(12, 1)])",
+          "\n      LogicalProject(col1=[$0], col2=[$1], EXPR$0=[$2], col10=[CAST($0):VARCHAR CHARACTER SET \"UTF-8\"], col20=[CAST($1):VARCHAR CHARACTER SET \"UTF-8\"], EXPR$05=[CAST($2):DECIMAL(12, 1)])",
           "\n        LogicalAggregate(group=[{0, 1}], agg#0=[$SUM0($2)])",
           "\n          PinotLogicalExchange(distribution=[hash[0, 1]])",
           "\n            LogicalAggregate(group=[{0, 1}], agg#0=[$SUM0($2)])",
@@ -359,7 +359,7 @@
           "\n    PinotLogicalExchange(distribution=[hash[0, 1]])",
           "\n      LogicalTableScan(table=[[a]])",
           "\n    PinotLogicalExchange(distribution=[hash[0, 1]])",
-          "\n      LogicalProject(col1=[$0], col2=[$1], EXPR$0=[CAST(/($2, $3)):DECIMAL(12, 1) NOT NULL], col10=[CAST($0):VARCHAR], col20=[CAST($1):VARCHAR], EXPR$05=[CAST(CAST(/($2, $3)):DECIMAL(12, 1) NOT NULL):DECIMAL(12, 1)])",
+          "\n      LogicalProject(col1=[$0], col2=[$1], EXPR$0=[CAST(/($2, $3)):DECIMAL(12, 1) NOT NULL], col10=[CAST($0):VARCHAR CHARACTER SET \"UTF-8\"], col20=[CAST($1):VARCHAR CHARACTER SET \"UTF-8\"], EXPR$05=[CAST(CAST(/($2, $3)):DECIMAL(12, 1) NOT NULL):DECIMAL(12, 1)])",
           "\n        LogicalAggregate(group=[{0, 1}], agg#0=[$SUM0($2)], agg#1=[COUNT($3)])",
           "\n          PinotLogicalExchange(distribution=[hash[0, 1]])",
           "\n            LogicalAggregate(group=[{0, 1}], agg#0=[$SUM0($2)], agg#1=[COUNT()])",
@@ -378,7 +378,7 @@
           "\n    PinotLogicalExchange(distribution=[hash[0, 1]])",
           "\n      LogicalTableScan(table=[[a]])",
           "\n    PinotLogicalExchange(distribution=[hash[0, 1]])",
-          "\n      LogicalProject(col1=[$0], col2=[$1], col10=[CAST($0):VARCHAR], col20=[CAST($1):VARCHAR], $f2=[CAST(/(CAST($2):DOUBLE NOT NULL, $3)):DOUBLE], EXPR$3=[*(0.5:DECIMAL(2, 1), /(CAST($2):DOUBLE NOT NULL, $3))])",
+          "\n      LogicalProject(col1=[$0], col2=[$1], col10=[CAST($0):VARCHAR CHARACTER SET \"UTF-8\"], col20=[CAST($1):VARCHAR CHARACTER SET \"UTF-8\"], $f2=[CAST(/(CAST($2):DOUBLE NOT NULL, $3)):DOUBLE], EXPR$3=[*(0.5:DECIMAL(2, 1), /(CAST($2):DOUBLE NOT NULL, $3))])",
           "\n        LogicalAggregate(group=[{0, 1}], agg#0=[$SUM0($2)], agg#1=[COUNT($3)])",
           "\n          PinotLogicalExchange(distribution=[hash[0, 1]])",
           "\n            LogicalAggregate(group=[{0, 1}], agg#0=[$SUM0($2)], agg#1=[COUNT()])",
@@ -434,7 +434,7 @@
           "\n      LogicalTableScan(table=[[a]])",
           "\n      PinotLogicalExchange(distribution=[broadcast], relExchangeType=[PIPELINE_BREAKER])",
           "\n        LogicalProject(col1=[$0])",
-          "\n          LogicalFilter(condition=[AND(<>($1, 'bar'), <>($1, 'foo'))])",
+          "\n          LogicalFilter(condition=[AND(<>($1, _UTF-8'bar'), <>($1, _UTF-8'foo'))])",
           "\n            LogicalTableScan(table=[[a]])",
           "\n    PinotLogicalExchange(distribution=[broadcast], relExchangeType=[PIPELINE_BREAKER])",
           "\n      LogicalProject(col1=[$0])",
@@ -462,7 +462,7 @@
           "\n              LogicalTableScan(table=[[a]])",
           "\n            PinotLogicalExchange(distribution=[broadcast], relExchangeType=[PIPELINE_BREAKER])",
           "\n              LogicalProject(col1=[$0])",
-          "\n                LogicalFilter(condition=[AND(<>($1, 'bar'), <>($1, 'foo'))])",
+          "\n                LogicalFilter(condition=[AND(<>($1, _UTF-8'bar'), <>($1, _UTF-8'foo'))])",
           "\n                  LogicalTableScan(table=[[a]])",
           "\n        PinotLogicalExchange(distribution=[broadcast], relExchangeType=[PIPELINE_BREAKER])",
           "\n          LogicalProject(col3=[$2])",
@@ -488,7 +488,7 @@
           "\n                  LogicalTableScan(table=[[a]])",
           "\n                PinotLogicalExchange(distribution=[broadcast], relExchangeType=[PIPELINE_BREAKER])",
           "\n                  LogicalProject(col1=[$0])",
-          "\n                    LogicalFilter(condition=[AND(<>($1, 'bar'), <>($1, 'foo'))])",
+          "\n                    LogicalFilter(condition=[AND(<>($1, _UTF-8'bar'), <>($1, _UTF-8'foo'))])",
           "\n                      LogicalTableScan(table=[[a]])",
           "\n            PinotLogicalExchange(distribution=[broadcast], relExchangeType=[PIPELINE_BREAKER])",
           "\n              LogicalProject(col3=[$2])",
@@ -508,11 +508,11 @@
           "\n      LogicalTableScan(table=[[b]])",
           "\n    PinotLogicalExchange(distribution=[broadcast], relExchangeType=[PIPELINE_BREAKER])",
           "\n      LogicalProject(col1=[$0])",
-          "\n        LogicalFilter(condition=[AND(<>($1, 'bar'), <>($1, 'foo'))])",
+          "\n        LogicalFilter(condition=[AND(<>($1, _UTF-8'bar'), <>($1, _UTF-8'foo'))])",
           "\n          LogicalTableScan(table=[[a]])",
           "\n  PinotLogicalExchange(distribution=[broadcast], relExchangeType=[PIPELINE_BREAKER])",
           "\n    LogicalProject(col3=[$2])",
-          "\n      LogicalFilter(condition=[AND(<>($1, 'bar'), <>($1, 'foo'))])",
+          "\n      LogicalFilter(condition=[AND(<>($1, _UTF-8'bar'), <>($1, _UTF-8'foo'))])",
           "\n        LogicalTableScan(table=[[a]])",
           "\n"
         ]
diff --git a/pinot-query-planner/src/test/resources/queries/LiteralEvaluationPlans.json b/pinot-query-planner/src/test/resources/queries/LiteralEvaluationPlans.json
index 52393c1bb364..71793c78237a 100644
--- a/pinot-query-planner/src/test/resources/queries/LiteralEvaluationPlans.json
+++ b/pinot-query-planner/src/test/resources/queries/LiteralEvaluationPlans.json
@@ -66,7 +66,7 @@
         "sql": "EXPLAIN PLAN FOR SELECT * FROM a WHERE  dateTrunc('MONTH', ts) =  dateTrunc('MONTH', FROMDATETIME('2019-01-01 00:00:00', 'yyyy-MM-dd HH:mm:ss'))",
         "output": [
           "Execution Plan",
-          "\nLogicalFilter(condition=[=(DATETRUNC('MONTH', $6), 1546300800000)])",
+          "\nLogicalFilter(condition=[=(DATETRUNC(_UTF-8'MONTH', $6), 1546300800000)])",
           "\n  LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -76,7 +76,7 @@
         "sql": "EXPLAIN PLAN FOR SELECT upper('month') FROM a",
         "output": [
           "Execution Plan",
-          "\nLogicalProject(EXPR$0=['MONTH'])",
+          "\nLogicalProject(EXPR$0=[_UTF-8'MONTH'])",
           "\n  LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -86,7 +86,7 @@
         "sql": "EXPLAIN PLAN FOR SELECT concat('month', ' 1') FROM a",
         "output": [
           "Execution Plan",
-          "\nLogicalProject(EXPR$0=[CAST('month 1':VARCHAR):VARCHAR])",
+          "\nLogicalProject(EXPR$0=[CAST(_UTF-8'month 1':VARCHAR CHARACTER SET \"UTF-8\"):VARCHAR CHARACTER SET \"UTF-8\"])",
           "\n  LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -96,7 +96,7 @@
         "sql": "EXPLAIN PLAN FOR SELECT substr('month',2) FROM a",
         "output": [
           "Execution Plan",
-          "\nLogicalProject(EXPR$0=[CAST('nth':VARCHAR):VARCHAR])",
+          "\nLogicalProject(EXPR$0=[CAST(_UTF-8'nth':VARCHAR CHARACTER SET \"UTF-8\"):VARCHAR CHARACTER SET \"UTF-8\"])",
           "\n  LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -106,7 +106,7 @@
         "sql": "EXPLAIN PLAN FOR SELECT upper(lower(upper(substr('month',2)))) FROM a",
         "output": [
           "Execution Plan",
-          "\nLogicalProject(EXPR$0=[CAST('NTH':VARCHAR):VARCHAR])",
+          "\nLogicalProject(EXPR$0=[CAST(_UTF-8'NTH':VARCHAR CHARACTER SET \"UTF-8\"):VARCHAR CHARACTER SET \"UTF-8\"])",
           "\n  LogicalTableScan(table=[[a]])",
           "\n"
         ]
diff --git a/pinot-query-planner/src/test/resources/queries/PinotHintablePlans.json b/pinot-query-planner/src/test/resources/queries/PinotHintablePlans.json
index 3277241224bf..7e9ff074d145 100644
--- a/pinot-query-planner/src/test/resources/queries/PinotHintablePlans.json
+++ b/pinot-query-planner/src/test/resources/queries/PinotHintablePlans.json
@@ -31,7 +31,7 @@
           "\n    LogicalJoin(condition=[=($0, $1)], joinType=[inner])",
           "\n      PinotLogicalExchange(distribution=[hash[0]])",
           "\n        LogicalProject(col1=[$0])",
-          "\n          LogicalFilter(condition=[AND(>=($2, 0), =($1, 'a'))])",
+          "\n          LogicalFilter(condition=[AND(>=($2, 0), =($1, _UTF-8'a'))])",
           "\n            LogicalTableScan(table=[[a]])",
           "\n      PinotLogicalExchange(distribution=[hash[0]])",
           "\n        LogicalProject(col2=[$1], col3=[$2])",
@@ -141,7 +141,7 @@
           "\n  LogicalAggregate(group=[{0, 1}], EXPR$2=[$SUM0($2)])",
           "\n    PinotLogicalExchange(distribution=[hash[0, 1]])",
           "\n      LogicalProject(col1=[$0], col2=[$1], col3=[$2])",
-          "\n        LogicalFilter(condition=[AND(>=($2, 0), =($0, 'a'))])",
+          "\n        LogicalFilter(condition=[AND(>=($2, 0), =($0, _UTF-8'a'))])",
           "\n          LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -156,7 +156,7 @@
           "\n    LogicalAggregate(group=[{0}], EXPR$1=[COUNT()], EXPR$2=[$SUM0($1)], EXPR$3=[$SUM0($2)], agg#3=[MAX($1)], agg#4=[MIN($1)])",
           "\n      PinotLogicalExchange(distribution=[hash[0]])",
           "\n        LogicalProject(col2=[$1], col3=[$2], $f2=[CAST($0):DECIMAL(1000, 500) NOT NULL])",
-          "\n          LogicalFilter(condition=[AND(>=($2, 0), =($1, 'a'))])",
+          "\n          LogicalFilter(condition=[AND(>=($2, 0), =($1, _UTF-8'a'))])",
           "\n            LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -169,7 +169,7 @@
           "\nLogicalFilter(condition=[>($1, 10)])",
           "\n  LogicalAggregate(group=[{0}], EXPR$1=[COUNT()], EXPR$2=[$SUM0($1)], EXPR$3=[$SUM0($2)])",
           "\n    LogicalProject(col2=[$1], col3=[$2], $f2=[CAST($0):DECIMAL(1000, 500) NOT NULL])",
-          "\n      LogicalFilter(condition=[AND(>=($2, 0), =($1, 'a'))])",
+          "\n      LogicalFilter(condition=[AND(>=($2, 0), =($1, _UTF-8'a'))])",
           "\n        LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -181,7 +181,7 @@
           "Execution Plan",
           "\nLogicalAggregate(group=[{0}], EXPR$1=[COUNT()], EXPR$2=[$SUM0($1)], EXPR$3=[$SUM0($2)])",
           "\n  LogicalProject(col2=[$1], col3=[$2], $f2=[CAST($0):DECIMAL(1000, 500) NOT NULL])",
-          "\n    LogicalFilter(condition=[AND(>=($2, 0), =($1, 'a'))])",
+          "\n    LogicalFilter(condition=[AND(>=($2, 0), =($1, _UTF-8'a'))])",
           "\n      LogicalTableScan(table=[[a]])",
           "\n"
         ]
diff --git a/pinot-query-planner/src/test/resources/queries/WindowFunctionPlans.json b/pinot-query-planner/src/test/resources/queries/WindowFunctionPlans.json
index fc04b684f4d7..076868fedbe8 100644
--- a/pinot-query-planner/src/test/resources/queries/WindowFunctionPlans.json
+++ b/pinot-query-planner/src/test/resources/queries/WindowFunctionPlans.json
@@ -179,7 +179,7 @@
           "\nLogicalProject($0=[$1], $1=[$2])",
           "\n  LogicalWindow(window#0=[window(aggs [SUM($0)])])",
           "\n    PinotLogicalExchange(distribution=[hash])",
-          "\n      LogicalProject(col3=[$2], $1=[CONCAT($0, '-', $1)])",
+          "\n      LogicalProject(col3=[$2], $1=[CONCAT($0, _UTF-8'-', $1)])",
           "\n        LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -219,8 +219,8 @@
           "\nLogicalProject($0=[$1], $1=[$2])",
           "\n  LogicalWindow(window#0=[window(aggs [MIN($0)])])",
           "\n    PinotLogicalExchange(distribution=[hash])",
-          "\n      LogicalProject(col3=[$2], $1=[CONCAT($0, '-', $1)])",
-          "\n        LogicalFilter(condition=[OR(=($0, 'bar'), =($0, 'foo'))])",
+          "\n      LogicalProject(col3=[$2], $1=[CONCAT($0, _UTF-8'-', $1)])",
+          "\n        LogicalFilter(condition=[OR(=($0, _UTF-8'bar'), =($0, _UTF-8'foo'))])",
           "\n          LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -434,7 +434,7 @@
           "\nLogicalProject($0=[$1], $1=[$2], $2=[$3])",
           "\n  LogicalWindow(window#0=[window(aggs [SUM($0), MAX($0)])])",
           "\n    PinotLogicalExchange(distribution=[hash])",
-          "\n      LogicalProject(col3=[$2], $1=[CONCAT($0, '-', $1)])",
+          "\n      LogicalProject(col3=[$2], $1=[CONCAT($0, _UTF-8'-', $1)])",
           "\n        LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -461,8 +461,8 @@
           "\nLogicalProject($0=[$1], $1=[$2], $2=[$3])",
           "\n  LogicalWindow(window#0=[window(aggs [MIN($0), MAX($0)])])",
           "\n    PinotLogicalExchange(distribution=[hash])",
-          "\n      LogicalProject(col3=[$2], $1=[LENGTH(CONCAT($0, ' ', $1))])",
-          "\n        LogicalFilter(condition=[AND(<>($0, 'bar'), <>($0, 'baz'), <>($0, 'foo'))])",
+          "\n      LogicalProject(col3=[$2], $1=[LENGTH(CONCAT($0, _UTF-8' ', $1))])",
+          "\n        LogicalFilter(condition=[AND(<>($0, _UTF-8'bar'), <>($0, _UTF-8'baz'), <>($0, _UTF-8'foo'))])",
           "\n          LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -476,8 +476,8 @@
           "\nLogicalProject(EXPR$0=[$0], EXPR$1=[$1], EXPR$2=[$1])",
           "\n  LogicalWindow(window#0=[window( rows between UNBOUNDED PRECEDING and CURRENT ROW aggs [ROW_NUMBER()])])",
           "\n    PinotLogicalExchange(distribution=[hash])",
-          "\n      LogicalProject($0=[LENGTH(CONCAT($0, ' ', $1))])",
-          "\n        LogicalFilter(condition=[AND(<>($0, 'bar'), <>($0, 'baz'), <>($0, 'foo'))])",
+          "\n      LogicalProject($0=[LENGTH(CONCAT($0, _UTF-8' ', $1))])",
+          "\n        LogicalFilter(condition=[AND(<>($0, _UTF-8'bar'), <>($0, _UTF-8'baz'), <>($0, _UTF-8'foo'))])",
           "\n          LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -768,8 +768,8 @@
           "\nLogicalProject(EXPR$0=[$2], EXPR$1=[/(CAST($3):DOUBLE NOT NULL, $4)])",
           "\n  LogicalWindow(window#0=[window(partition {0} aggs [SUM($1), COUNT($1)])])",
           "\n    PinotLogicalExchange(distribution=[hash[0]])",
-          "\n      LogicalProject(col2=[$1], col3=[$2], $2=[CONCAT($0, '-', $1)])",
-          "\n        LogicalFilter(condition=[OR(AND(<>($0, 'bar'), <>($0, 'foo')), >=($2, 42))])",
+          "\n      LogicalProject(col2=[$1], col3=[$2], $2=[CONCAT($0, _UTF-8'-', $1)])",
+          "\n        LogicalFilter(condition=[OR(AND(<>($0, _UTF-8'bar'), <>($0, _UTF-8'foo')), >=($2, 42))])",
           "\n          LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -782,7 +782,7 @@
           "\nLogicalProject(EXPR$0=[/(CAST($2):DOUBLE NOT NULL, $3)])",
           "\n  LogicalWindow(window#0=[window(partition {1} aggs [SUM($0), COUNT($0)])])",
           "\n    PinotLogicalExchange(distribution=[hash[1]])",
-          "\n      LogicalProject(col3=[$2], $1=[CONCAT($0, '-', $1)])",
+          "\n      LogicalProject(col3=[$2], $1=[CONCAT($0, _UTF-8'-', $1)])",
           "\n        LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -795,7 +795,7 @@
           "\nLogicalProject($0=[$1])",
           "\n  LogicalWindow(window#0=[window(partition {0} rows between UNBOUNDED PRECEDING and CURRENT ROW aggs [ROW_NUMBER()])])",
           "\n    PinotLogicalExchange(distribution=[hash[0]])",
-          "\n      LogicalProject($0=[CONCAT($0, '-', $1)])",
+          "\n      LogicalProject($0=[CONCAT($0, _UTF-8'-', $1)])",
           "\n        LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -1113,7 +1113,7 @@
           "\n  LogicalWindow(window#0=[window(partition {0} aggs [SUM($1), COUNT($1), COUNT($0)])])",
           "\n    PinotLogicalExchange(distribution=[hash[0]])",
           "\n      LogicalProject(col1=[$0], col3=[$2])",
-          "\n        LogicalFilter(condition=[AND(>($2, 42), OR(=($0, 'chewbacca':VARCHAR(9)), =($0, 'vader':VARCHAR(9)), =($0, 'yoda':VARCHAR(9))))])",
+          "\n        LogicalFilter(condition=[AND(>($2, 42), OR(=($0, _UTF-8'chewbacca':VARCHAR(9) CHARACTER SET \"UTF-8\"), =($0, _UTF-8'vader':VARCHAR(9) CHARACTER SET \"UTF-8\"), =($0, _UTF-8'yoda':VARCHAR(9) CHARACTER SET \"UTF-8\")))])",
           "\n          LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -1126,8 +1126,8 @@
           "\nLogicalProject($0=[$2], $1=[$3], $2=[$4])",
           "\n  LogicalWindow(window#0=[window(partition {0} aggs [MIN($1), MAX($1)])])",
           "\n    PinotLogicalExchange(distribution=[hash[0]])",
-          "\n      LogicalProject(col1=[$0], col3=[$2], $2=[REVERSE(CONCAT($0, ' ', $1))])",
-          "\n        LogicalFilter(condition=[AND(<>($1, 'bar'), <>($1, 'baz'), <>($1, 'foo'))])",
+          "\n      LogicalProject(col1=[$0], col3=[$2], $2=[REVERSE(CONCAT($0, _UTF-8' ', $1))])",
+          "\n        LogicalFilter(condition=[AND(<>($1, _UTF-8'bar'), <>($1, _UTF-8'baz'), <>($1, _UTF-8'foo'))])",
           "\n          LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -1140,7 +1140,7 @@
           "\nLogicalProject(EXPR$0=[/(CAST($3):DOUBLE NOT NULL, $4)], EXPR$1=[$5])",
           "\n  LogicalWindow(window#0=[window(partition {2} aggs [SUM($1), COUNT($1), COUNT($0)])])",
           "\n    PinotLogicalExchange(distribution=[hash[2]])",
-          "\n      LogicalProject(col1=[$0], col3=[$2], $2=[REVERSE(CONCAT($0, '-', $1))])",
+          "\n      LogicalProject(col1=[$0], col3=[$2], $2=[REVERSE(CONCAT($0, _UTF-8'-', $1))])",
           "\n        LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -1627,8 +1627,8 @@
           "\nLogicalProject(EXPR$0=[$2], EXPR$1=[/(CAST($3):DOUBLE NOT NULL, $4)])",
           "\n  LogicalWindow(window#0=[window(order by [0] aggs [SUM($1), COUNT($1)])])",
           "\n    PinotLogicalSortExchange(distribution=[hash], collation=[[0]], isSortOnSender=[false], isSortOnReceiver=[true])",
-          "\n      LogicalProject(col2=[$1], col3=[$2], $2=[CONCAT($0, '-', $1)])",
-          "\n        LogicalFilter(condition=[OR(AND(<>($0, 'bar'), <>($0, 'foo')), >=($2, 42))])",
+          "\n      LogicalProject(col2=[$1], col3=[$2], $2=[CONCAT($0, _UTF-8'-', $1)])",
+          "\n        LogicalFilter(condition=[OR(AND(<>($0, _UTF-8'bar'), <>($0, _UTF-8'foo')), >=($2, 42))])",
           "\n          LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -1641,8 +1641,8 @@
           "\nLogicalProject($0=[$1], $1=[$2])",
           "\n  LogicalWindow(window#0=[window(order by [0] rows between UNBOUNDED PRECEDING and CURRENT ROW aggs [ROW_NUMBER()])])",
           "\n    PinotLogicalSortExchange(distribution=[hash], collation=[[0]], isSortOnSender=[false], isSortOnReceiver=[true])",
-          "\n      LogicalProject(col2=[$1], $1=[CONCAT($0, '-', $1)])",
-          "\n        LogicalFilter(condition=[OR(AND(<>($0, 'bar'), <>($0, 'foo')), >=($2, 42))])",
+          "\n      LogicalProject(col2=[$1], $1=[CONCAT($0, _UTF-8'-', $1)])",
+          "\n        LogicalFilter(condition=[OR(AND(<>($0, _UTF-8'bar'), <>($0, _UTF-8'foo')), >=($2, 42))])",
           "\n          LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -1655,8 +1655,8 @@
           "\nLogicalProject($0=[$1], $1=[$2])",
           "\n  LogicalWindow(window#0=[window(order by [0] aggs [RANK()])])",
           "\n    PinotLogicalSortExchange(distribution=[hash], collation=[[0]], isSortOnSender=[false], isSortOnReceiver=[true])",
-          "\n      LogicalProject(col2=[$1], $1=[CONCAT($0, '-', $1)])",
-          "\n        LogicalFilter(condition=[OR(AND(<>($0, 'bar'), <>($0, 'foo')), >=($2, 42))])",
+          "\n      LogicalProject(col2=[$1], $1=[CONCAT($0, _UTF-8'-', $1)])",
+          "\n        LogicalFilter(condition=[OR(AND(<>($0, _UTF-8'bar'), <>($0, _UTF-8'foo')), >=($2, 42))])",
           "\n          LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -1669,7 +1669,7 @@
           "\nLogicalProject(EXPR$0=[/(CAST($2):DOUBLE NOT NULL, $3)])",
           "\n  LogicalWindow(window#0=[window(order by [1] aggs [SUM($0), COUNT($0)])])",
           "\n    PinotLogicalSortExchange(distribution=[hash], collation=[[1]], isSortOnSender=[false], isSortOnReceiver=[true])",
-          "\n      LogicalProject(col3=[$2], $1=[CONCAT($0, '-', $1)])",
+          "\n      LogicalProject(col3=[$2], $1=[CONCAT($0, _UTF-8'-', $1)])",
           "\n        LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -1682,7 +1682,7 @@
           "\nLogicalProject($0=[$1])",
           "\n  LogicalWindow(window#0=[window(order by [0] rows between UNBOUNDED PRECEDING and CURRENT ROW aggs [ROW_NUMBER()])])",
           "\n    PinotLogicalSortExchange(distribution=[hash], collation=[[0]], isSortOnSender=[false], isSortOnReceiver=[true])",
-          "\n      LogicalProject($0=[CONCAT($0, '-', $1)])",
+          "\n      LogicalProject($0=[CONCAT($0, _UTF-8'-', $1)])",
           "\n        LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -1695,7 +1695,7 @@
           "\nLogicalProject($0=[$1])",
           "\n  LogicalWindow(window#0=[window(order by [0] aggs [DENSE_RANK()])])",
           "\n    PinotLogicalSortExchange(distribution=[hash], collation=[[0]], isSortOnSender=[false], isSortOnReceiver=[true])",
-          "\n      LogicalProject($0=[CONCAT($0, '-', $1)])",
+          "\n      LogicalProject($0=[CONCAT($0, _UTF-8'-', $1)])",
           "\n        LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -1947,7 +1947,7 @@
           "\n  LogicalWindow(window#0=[window(order by [0] aggs [SUM($1), COUNT($1), COUNT($0)])])",
           "\n    PinotLogicalSortExchange(distribution=[hash], collation=[[0]], isSortOnSender=[false], isSortOnReceiver=[true])",
           "\n      LogicalProject(col1=[$0], col3=[$2])",
-          "\n        LogicalFilter(condition=[AND(>($2, 42), OR(=($0, 'chewbacca':VARCHAR(9)), =($0, 'vader':VARCHAR(9)), =($0, 'yoda':VARCHAR(9))))])",
+          "\n        LogicalFilter(condition=[AND(>($2, 42), OR(=($0, _UTF-8'chewbacca':VARCHAR(9) CHARACTER SET \"UTF-8\"), =($0, _UTF-8'vader':VARCHAR(9) CHARACTER SET \"UTF-8\"), =($0, _UTF-8'yoda':VARCHAR(9) CHARACTER SET \"UTF-8\")))])",
           "\n          LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -1960,8 +1960,8 @@
           "\nLogicalProject($0=[$2], $1=[$3], $2=[$4])",
           "\n  LogicalWindow(window#0=[window(order by [0] aggs [MIN($1), MAX($1)])])",
           "\n    PinotLogicalSortExchange(distribution=[hash], collation=[[0]], isSortOnSender=[false], isSortOnReceiver=[true])",
-          "\n      LogicalProject(col1=[$0], col3=[$2], $2=[REVERSE(CONCAT($0, ' ', $1))])",
-          "\n        LogicalFilter(condition=[AND(<>($1, 'bar'), <>($1, 'baz'), <>($1, 'foo'))])",
+          "\n      LogicalProject(col1=[$0], col3=[$2], $2=[REVERSE(CONCAT($0, _UTF-8' ', $1))])",
+          "\n        LogicalFilter(condition=[AND(<>($1, _UTF-8'bar'), <>($1, _UTF-8'baz'), <>($1, _UTF-8'foo'))])",
           "\n          LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -1974,8 +1974,8 @@
           "\nLogicalProject($0=[$1], $1=[$2], $2=[$3])",
           "\n  LogicalWindow(window#0=[window(order by [0] aggs [RANK(), DENSE_RANK()])])",
           "\n    PinotLogicalSortExchange(distribution=[hash], collation=[[0]], isSortOnSender=[false], isSortOnReceiver=[true])",
-          "\n      LogicalProject(col1=[$0], $1=[REVERSE(CONCAT($0, ' ', $1))])",
-          "\n        LogicalFilter(condition=[AND(<>($1, 'bar'), <>($1, 'baz'), <>($1, 'foo'))])",
+          "\n      LogicalProject(col1=[$0], $1=[REVERSE(CONCAT($0, _UTF-8' ', $1))])",
+          "\n        LogicalFilter(condition=[AND(<>($1, _UTF-8'bar'), <>($1, _UTF-8'baz'), <>($1, _UTF-8'foo'))])",
           "\n          LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -1988,7 +1988,7 @@
           "\nLogicalProject(EXPR$0=[/(CAST($3):DOUBLE NOT NULL, $4)], EXPR$1=[$5])",
           "\n  LogicalWindow(window#0=[window(order by [2] aggs [SUM($1), COUNT($1), COUNT($0)])])",
           "\n    PinotLogicalSortExchange(distribution=[hash], collation=[[2]], isSortOnSender=[false], isSortOnReceiver=[true])",
-          "\n      LogicalProject(col1=[$0], col3=[$2], $2=[REVERSE(CONCAT($0, '-', $1))])",
+          "\n      LogicalProject(col1=[$0], col3=[$2], $2=[REVERSE(CONCAT($0, _UTF-8'-', $1))])",
           "\n        LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -2001,7 +2001,7 @@
           "\nLogicalProject($0=[$2], $1=[$3])",
           "\n  LogicalWindow(window#0=[window(order by [1] aggs [DENSE_RANK(), COUNT($0)])])",
           "\n    PinotLogicalSortExchange(distribution=[hash], collation=[[1]], isSortOnSender=[false], isSortOnReceiver=[true])",
-          "\n      LogicalProject(col1=[$0], $1=[REVERSE(CONCAT($0, '-', $1))])",
+          "\n      LogicalProject(col1=[$0], $1=[REVERSE(CONCAT($0, _UTF-8'-', $1))])",
           "\n        LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -2296,8 +2296,8 @@
           "\nLogicalProject(EXPR$0=[$2], EXPR$1=[/(CAST($3):DOUBLE NOT NULL, $4)])",
           "\n  LogicalWindow(window#0=[window(partition {0} order by [0] aggs [SUM($1), COUNT($1)])])",
           "\n    PinotLogicalExchange(distribution=[hash[0]])",
-          "\n      LogicalProject(col2=[$1], col3=[$2], $2=[CONCAT($0, '-', $1)])",
-          "\n        LogicalFilter(condition=[OR(AND(<>($0, 'bar'), <>($0, 'foo')), >=($2, 42))])",
+          "\n      LogicalProject(col2=[$1], col3=[$2], $2=[CONCAT($0, _UTF-8'-', $1)])",
+          "\n        LogicalFilter(condition=[OR(AND(<>($0, _UTF-8'bar'), <>($0, _UTF-8'foo')), >=($2, 42))])",
           "\n          LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -2310,8 +2310,8 @@
           "\nLogicalProject($0=[$1], $1=[$2])",
           "\n  LogicalWindow(window#0=[window(partition {0} order by [0] aggs [RANK()])])",
           "\n    PinotLogicalExchange(distribution=[hash[0]])",
-          "\n      LogicalProject(col2=[$1], $1=[CONCAT($0, '-', $1)])",
-          "\n        LogicalFilter(condition=[OR(AND(<>($0, 'bar'), <>($0, 'foo')), >=($2, 42))])",
+          "\n      LogicalProject(col2=[$1], $1=[CONCAT($0, _UTF-8'-', $1)])",
+          "\n        LogicalFilter(condition=[OR(AND(<>($0, _UTF-8'bar'), <>($0, _UTF-8'foo')), >=($2, 42))])",
           "\n          LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -2324,7 +2324,7 @@
           "\nLogicalProject(EXPR$0=[/(CAST($2):DOUBLE NOT NULL, $3)])",
           "\n  LogicalWindow(window#0=[window(partition {1} order by [1] aggs [SUM($0), COUNT($0)])])",
           "\n    PinotLogicalExchange(distribution=[hash[1]])",
-          "\n      LogicalProject(col3=[$2], $1=[CONCAT($0, '-', $1)])",
+          "\n      LogicalProject(col3=[$2], $1=[CONCAT($0, _UTF-8'-', $1)])",
           "\n        LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -2337,7 +2337,7 @@
           "\nLogicalProject($0=[$1])",
           "\n  LogicalWindow(window#0=[window(partition {0} order by [0] aggs [DENSE_RANK()])])",
           "\n    PinotLogicalExchange(distribution=[hash[0]])",
-          "\n      LogicalProject($0=[CONCAT($0, '-', $1)])",
+          "\n      LogicalProject($0=[CONCAT($0, _UTF-8'-', $1)])",
           "\n        LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -2600,7 +2600,7 @@
           "\n  LogicalWindow(window#0=[window(partition {0} order by [0] aggs [SUM($1), COUNT($1), COUNT($0)])])",
           "\n    PinotLogicalExchange(distribution=[hash[0]])",
           "\n      LogicalProject(col1=[$0], col3=[$2])",
-          "\n        LogicalFilter(condition=[AND(>($2, 42), OR(=($0, 'chewbacca':VARCHAR(9)), =($0, 'vader':VARCHAR(9)), =($0, 'yoda':VARCHAR(9))))])",
+          "\n        LogicalFilter(condition=[AND(>($2, 42), OR(=($0, _UTF-8'chewbacca':VARCHAR(9) CHARACTER SET \"UTF-8\"), =($0, _UTF-8'vader':VARCHAR(9) CHARACTER SET \"UTF-8\"), =($0, _UTF-8'yoda':VARCHAR(9) CHARACTER SET \"UTF-8\")))])",
           "\n          LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -2613,7 +2613,7 @@
           "\nLogicalWindow(window#0=[window(partition {0} order by [0] aggs [RANK(), DENSE_RANK()])])",
           "\n  PinotLogicalExchange(distribution=[hash[0]])",
           "\n    LogicalProject(col1=[$0])",
-          "\n      LogicalFilter(condition=[AND(>($2, 42), OR(=($0, 'chewbacca':VARCHAR(9)), =($0, 'vader':VARCHAR(9)), =($0, 'yoda':VARCHAR(9))))])",
+          "\n      LogicalFilter(condition=[AND(>($2, 42), OR(=($0, _UTF-8'chewbacca':VARCHAR(9) CHARACTER SET \"UTF-8\"), =($0, _UTF-8'vader':VARCHAR(9) CHARACTER SET \"UTF-8\"), =($0, _UTF-8'yoda':VARCHAR(9) CHARACTER SET \"UTF-8\")))])",
           "\n        LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -2626,8 +2626,8 @@
           "\nLogicalProject($0=[$2], $1=[$3], $2=[$4])",
           "\n  LogicalWindow(window#0=[window(partition {0} order by [0] aggs [MIN($1), MAX($1)])])",
           "\n    PinotLogicalExchange(distribution=[hash[0]])",
-          "\n      LogicalProject(col1=[$0], col3=[$2], $2=[REVERSE(CONCAT($0, ' ', $1))])",
-          "\n        LogicalFilter(condition=[AND(<>($1, 'bar'), <>($1, 'baz'), <>($1, 'foo'))])",
+          "\n      LogicalProject(col1=[$0], col3=[$2], $2=[REVERSE(CONCAT($0, _UTF-8' ', $1))])",
+          "\n        LogicalFilter(condition=[AND(<>($1, _UTF-8'bar'), <>($1, _UTF-8'baz'), <>($1, _UTF-8'foo'))])",
           "\n          LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -2640,8 +2640,8 @@
           "\nLogicalProject($0=[$2], $1=[$3], $2=[$4])",
           "\n  LogicalWindow(window#0=[window(partition {0} order by [0] aggs [DENSE_RANK(), MAX($1)])])",
           "\n    PinotLogicalExchange(distribution=[hash[0]])",
-          "\n      LogicalProject(col1=[$0], col3=[$2], $2=[REVERSE(CONCAT($0, ' ', $1))])",
-          "\n        LogicalFilter(condition=[AND(<>($1, 'bar'), <>($1, 'baz'), <>($1, 'foo'))])",
+          "\n      LogicalProject(col1=[$0], col3=[$2], $2=[REVERSE(CONCAT($0, _UTF-8' ', $1))])",
+          "\n        LogicalFilter(condition=[AND(<>($1, _UTF-8'bar'), <>($1, _UTF-8'baz'), <>($1, _UTF-8'foo'))])",
           "\n          LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -2654,7 +2654,7 @@
           "\nLogicalProject(EXPR$0=[/(CAST($3):DOUBLE NOT NULL, $4)], EXPR$1=[$5])",
           "\n  LogicalWindow(window#0=[window(partition {2} order by [2] aggs [SUM($1), COUNT($1), COUNT($0)])])",
           "\n    PinotLogicalExchange(distribution=[hash[2]])",
-          "\n      LogicalProject(col1=[$0], col3=[$2], $2=[REVERSE(CONCAT($0, '-', $1))])",
+          "\n      LogicalProject(col1=[$0], col3=[$2], $2=[REVERSE(CONCAT($0, _UTF-8'-', $1))])",
           "\n        LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -2668,7 +2668,7 @@
           "\nLogicalProject(EXPR$0=[$1], EXPR$1=[$1])",
           "\n  LogicalWindow(window#0=[window(partition {0} order by [0] rows between UNBOUNDED PRECEDING and CURRENT ROW aggs [ROW_NUMBER()])])",
           "\n    PinotLogicalExchange(distribution=[hash[0]])",
-          "\n      LogicalProject($0=[REVERSE(CONCAT($0, '-', $1))])",
+          "\n      LogicalProject($0=[REVERSE(CONCAT($0, _UTF-8'-', $1))])",
           "\n        LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -2681,7 +2681,7 @@
           "\nLogicalProject($0=[$1], $1=[$2])",
           "\n  LogicalWindow(window#0=[window(partition {0} order by [0] aggs [RANK(), DENSE_RANK()])])",
           "\n    PinotLogicalExchange(distribution=[hash[0]])",
-          "\n      LogicalProject($0=[REVERSE(CONCAT($0, '-', $1))])",
+          "\n      LogicalProject($0=[REVERSE(CONCAT($0, _UTF-8'-', $1))])",
           "\n        LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -2987,8 +2987,8 @@
           "\nLogicalProject(EXPR$0=[$3], EXPR$1=[/(CAST($4):DOUBLE NOT NULL, $5)])",
           "\n  LogicalWindow(window#0=[window(partition {1} order by [0] aggs [SUM($2), COUNT($2)])])",
           "\n    PinotLogicalSortExchange(distribution=[hash[1]], collation=[[0]], isSortOnSender=[false], isSortOnReceiver=[true])",
-          "\n      LogicalProject(col1=[$0], col2=[$1], col3=[$2], $3=[CONCAT($0, '-', $1)])",
-          "\n        LogicalFilter(condition=[OR(AND(<>($0, 'bar'), <>($0, 'foo')), >=($2, 42))])",
+          "\n      LogicalProject(col1=[$0], col2=[$1], col3=[$2], $3=[CONCAT($0, _UTF-8'-', $1)])",
+          "\n        LogicalFilter(condition=[OR(AND(<>($0, _UTF-8'bar'), <>($0, _UTF-8'foo')), >=($2, 42))])",
           "\n          LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -3001,8 +3001,8 @@
           "\nLogicalProject($0=[$2], $1=[$3])",
           "\n  LogicalWindow(window#0=[window(partition {1} order by [0] rows between UNBOUNDED PRECEDING and CURRENT ROW aggs [ROW_NUMBER()])])",
           "\n    PinotLogicalSortExchange(distribution=[hash[1]], collation=[[0]], isSortOnSender=[false], isSortOnReceiver=[true])",
-          "\n      LogicalProject(col1=[$0], col2=[$1], $2=[CONCAT($0, '-', $1)])",
-          "\n        LogicalFilter(condition=[OR(AND(<>($0, 'bar'), <>($0, 'foo')), >=($2, 42))])",
+          "\n      LogicalProject(col1=[$0], col2=[$1], $2=[CONCAT($0, _UTF-8'-', $1)])",
+          "\n        LogicalFilter(condition=[OR(AND(<>($0, _UTF-8'bar'), <>($0, _UTF-8'foo')), >=($2, 42))])",
           "\n          LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -3015,8 +3015,8 @@
           "\nLogicalProject($0=[$2], $1=[$3])",
           "\n  LogicalWindow(window#0=[window(partition {1} order by [0] aggs [RANK()])])",
           "\n    PinotLogicalSortExchange(distribution=[hash[1]], collation=[[0]], isSortOnSender=[false], isSortOnReceiver=[true])",
-          "\n      LogicalProject(col1=[$0], col2=[$1], $2=[CONCAT($0, '-', $1)])",
-          "\n        LogicalFilter(condition=[OR(AND(<>($0, 'bar'), <>($0, 'foo')), >=($2, 42))])",
+          "\n      LogicalProject(col1=[$0], col2=[$1], $2=[CONCAT($0, _UTF-8'-', $1)])",
+          "\n        LogicalFilter(condition=[OR(AND(<>($0, _UTF-8'bar'), <>($0, _UTF-8'foo')), >=($2, 42))])",
           "\n          LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -3029,7 +3029,7 @@
           "\nLogicalProject(EXPR$0=[/(CAST($3):DOUBLE NOT NULL, $4)])",
           "\n  LogicalWindow(window#0=[window(partition {2} order by [1] aggs [SUM($0), COUNT($0)])])",
           "\n    PinotLogicalSortExchange(distribution=[hash[2]], collation=[[1]], isSortOnSender=[false], isSortOnReceiver=[true])",
-          "\n      LogicalProject(col3=[$2], $1=[REVERSE($1)], $2=[CONCAT($0, '-', $1)])",
+          "\n      LogicalProject(col3=[$2], $1=[REVERSE($1)], $2=[CONCAT($0, _UTF-8'-', $1)])",
           "\n        LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -3042,7 +3042,7 @@
           "\nLogicalProject($0=[$2])",
           "\n  LogicalWindow(window#0=[window(partition {1} order by [0] rows between UNBOUNDED PRECEDING and CURRENT ROW aggs [ROW_NUMBER()])])",
           "\n    PinotLogicalSortExchange(distribution=[hash[1]], collation=[[0]], isSortOnSender=[false], isSortOnReceiver=[true])",
-          "\n      LogicalProject($0=[REVERSE($1)], $1=[CONCAT($0, '-', $1)])",
+          "\n      LogicalProject($0=[REVERSE($1)], $1=[CONCAT($0, _UTF-8'-', $1)])",
           "\n        LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -3055,7 +3055,7 @@
           "\nLogicalProject($0=[$2])",
           "\n  LogicalWindow(window#0=[window(partition {1} order by [0] aggs [DENSE_RANK()])])",
           "\n    PinotLogicalSortExchange(distribution=[hash[1]], collation=[[0]], isSortOnSender=[false], isSortOnReceiver=[true])",
-          "\n      LogicalProject($0=[REVERSE($1)], $1=[CONCAT($0, '-', $1)])",
+          "\n      LogicalProject($0=[REVERSE($1)], $1=[CONCAT($0, _UTF-8'-', $1)])",
           "\n        LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -3267,7 +3267,7 @@
           "\n  LogicalWindow(window#0=[window(partition {0} order by [1] aggs [SUM($2), COUNT($2), COUNT($0)])])",
           "\n    PinotLogicalSortExchange(distribution=[hash[0]], collation=[[1]], isSortOnSender=[false], isSortOnReceiver=[true])",
           "\n      LogicalProject(col1=[$0], col2=[$1], col3=[$2])",
-          "\n        LogicalFilter(condition=[AND(>($2, 42), OR(=($0, 'chewbacca':VARCHAR(9)), =($0, 'vader':VARCHAR(9)), =($0, 'yoda':VARCHAR(9))))])",
+          "\n        LogicalFilter(condition=[AND(>($2, 42), OR(=($0, _UTF-8'chewbacca':VARCHAR(9) CHARACTER SET \"UTF-8\"), =($0, _UTF-8'vader':VARCHAR(9) CHARACTER SET \"UTF-8\"), =($0, _UTF-8'yoda':VARCHAR(9) CHARACTER SET \"UTF-8\")))])",
           "\n          LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -3281,7 +3281,7 @@
           "\n  LogicalWindow(window#0=[window(partition {0} order by [1] rows between UNBOUNDED PRECEDING and CURRENT ROW aggs [ROW_NUMBER()])])",
           "\n    PinotLogicalSortExchange(distribution=[hash[0]], collation=[[1]], isSortOnSender=[false], isSortOnReceiver=[true])",
           "\n      LogicalProject(col1=[$0], col2=[$1])",
-          "\n        LogicalFilter(condition=[AND(>($2, 42), OR(=($0, 'chewbacca':VARCHAR(9)), =($0, 'vader':VARCHAR(9)), =($0, 'yoda':VARCHAR(9))))])",
+          "\n        LogicalFilter(condition=[AND(>($2, 42), OR(=($0, _UTF-8'chewbacca':VARCHAR(9) CHARACTER SET \"UTF-8\"), =($0, _UTF-8'vader':VARCHAR(9) CHARACTER SET \"UTF-8\"), =($0, _UTF-8'yoda':VARCHAR(9) CHARACTER SET \"UTF-8\")))])",
           "\n          LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -3295,7 +3295,7 @@
           "\n  LogicalWindow(window#0=[window(partition {0} order by [1] aggs [RANK(), DENSE_RANK()])])",
           "\n    PinotLogicalSortExchange(distribution=[hash[0]], collation=[[1]], isSortOnSender=[false], isSortOnReceiver=[true])",
           "\n      LogicalProject(col1=[$0], col2=[$1])",
-          "\n        LogicalFilter(condition=[AND(>($2, 42), OR(=($0, 'chewbacca':VARCHAR(9)), =($0, 'vader':VARCHAR(9)), =($0, 'yoda':VARCHAR(9))))])",
+          "\n        LogicalFilter(condition=[AND(>($2, 42), OR(=($0, _UTF-8'chewbacca':VARCHAR(9) CHARACTER SET \"UTF-8\"), =($0, _UTF-8'vader':VARCHAR(9) CHARACTER SET \"UTF-8\"), =($0, _UTF-8'yoda':VARCHAR(9) CHARACTER SET \"UTF-8\")))])",
           "\n          LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -3308,8 +3308,8 @@
           "\nLogicalProject($0=[$3], $1=[$4], $2=[$5])",
           "\n  LogicalWindow(window#0=[window(partition {0} order by [1] aggs [MIN($2), MAX($2)])])",
           "\n    PinotLogicalSortExchange(distribution=[hash[0]], collation=[[1]], isSortOnSender=[false], isSortOnReceiver=[true])",
-          "\n      LogicalProject(col1=[$0], col2=[$1], col3=[$2], $3=[REVERSE(CONCAT($0, ' ', $1))])",
-          "\n        LogicalFilter(condition=[AND(<>($1, 'bar'), <>($1, 'baz'), <>($1, 'foo'))])",
+          "\n      LogicalProject(col1=[$0], col2=[$1], col3=[$2], $3=[REVERSE(CONCAT($0, _UTF-8' ', $1))])",
+          "\n        LogicalFilter(condition=[AND(<>($1, _UTF-8'bar'), <>($1, _UTF-8'baz'), <>($1, _UTF-8'foo'))])",
           "\n          LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -3322,8 +3322,8 @@
           "\nLogicalProject($0=[$2], $1=[$3], $2=[$4])",
           "\n  LogicalWindow(window#0=[window(partition {0} order by [1] aggs [DENSE_RANK(), RANK()])])",
           "\n    PinotLogicalSortExchange(distribution=[hash[0]], collation=[[1]], isSortOnSender=[false], isSortOnReceiver=[true])",
-          "\n      LogicalProject(col1=[$0], col2=[$1], $2=[REVERSE(CONCAT($0, ' ', $1))])",
-          "\n        LogicalFilter(condition=[AND(<>($1, 'bar'), <>($1, 'baz'), <>($1, 'foo'))])",
+          "\n      LogicalProject(col1=[$0], col2=[$1], $2=[REVERSE(CONCAT($0, _UTF-8' ', $1))])",
+          "\n        LogicalFilter(condition=[AND(<>($1, _UTF-8'bar'), <>($1, _UTF-8'baz'), <>($1, _UTF-8'foo'))])",
           "\n          LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -3336,7 +3336,7 @@
           "\nLogicalProject(EXPR$0=[/(CAST($4):DOUBLE NOT NULL, $5)], EXPR$1=[$6])",
           "\n  LogicalWindow(window#0=[window(partition {3} order by [2] aggs [SUM($1), COUNT($1), COUNT($0)])])",
           "\n    PinotLogicalSortExchange(distribution=[hash[3]], collation=[[2]], isSortOnSender=[false], isSortOnReceiver=[true])",
-          "\n      LogicalProject(col1=[$0], col3=[$2], $2=[CONCAT($0, '-', $1)], $3=[REVERSE(CONCAT($0, '-', $1))])",
+          "\n      LogicalProject(col1=[$0], col3=[$2], $2=[CONCAT($0, _UTF-8'-', $1)], $3=[REVERSE(CONCAT($0, _UTF-8'-', $1))])",
           "\n        LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -3349,7 +3349,7 @@
           "\nLogicalProject(EXPR$0=[/(CAST($3):DOUBLE NOT NULL, $4)], EXPR$1=[$5])",
           "\n  LogicalWindow(window#0=[window(partition {2} order by [1] aggs [SUM($0), COUNT($0), RANK()])])",
           "\n    PinotLogicalSortExchange(distribution=[hash[2]], collation=[[1]], isSortOnSender=[false], isSortOnReceiver=[true])",
-          "\n      LogicalProject(col3=[$2], $1=[CONCAT($0, '-', $1)], $2=[REVERSE(CONCAT($0, '-', $1))])",
+          "\n      LogicalProject(col3=[$2], $1=[CONCAT($0, _UTF-8'-', $1)], $2=[REVERSE(CONCAT($0, _UTF-8'-', $1))])",
           "\n        LogicalTableScan(table=[[a]])",
           "\n"
         ]
@@ -3507,7 +3507,7 @@
           "\n                  LogicalTableScan(table=[[a]])",
           "\n            PinotLogicalExchange(distribution=[hash[0]])",
           "\n              LogicalProject(col2=[$1])",
-          "\n                LogicalFilter(condition=[OR(=($0, 'brandon sanderson':VARCHAR(17)), =($0, 'douglas adams':VARCHAR(17)))])",
+          "\n                LogicalFilter(condition=[OR(=($0, _UTF-8'brandon sanderson':VARCHAR(17) CHARACTER SET \"UTF-8\"), =($0, _UTF-8'douglas adams':VARCHAR(17) CHARACTER SET \"UTF-8\"))])",
           "\n                  LogicalTableScan(table=[[b]])",
           "\n"
         ]
@@ -3529,7 +3529,7 @@
           "\n                  LogicalTableScan(table=[[a]])",
           "\n            PinotLogicalExchange(distribution=[hash[0]])",
           "\n              LogicalProject(col2=[$1])",
-          "\n                LogicalFilter(condition=[OR(=($0, 'brandon sanderson':VARCHAR(17)), =($0, 'douglas adams':VARCHAR(17)))])",
+          "\n                LogicalFilter(condition=[OR(=($0, _UTF-8'brandon sanderson':VARCHAR(17) CHARACTER SET \"UTF-8\"), =($0, _UTF-8'douglas adams':VARCHAR(17) CHARACTER SET \"UTF-8\"))])",
           "\n                  LogicalTableScan(table=[[b]])",
           "\n"
         ]
diff --git a/pinot-query-runtime/src/test/resources/queries/CharacterTypes.json b/pinot-query-runtime/src/test/resources/queries/CharacterTypes.json
index ffe4c3b0bb65..2573d980e10b 100644
--- a/pinot-query-runtime/src/test/resources/queries/CharacterTypes.json
+++ b/pinot-query-runtime/src/test/resources/queries/CharacterTypes.json
@@ -32,8 +32,6 @@
         "sql": "SELECT 'value with spaces' FROM {varchar}"
       },
       {
-        "ignored": true,
-        "comment": "calcite doesn't support parsing non ISO-8859-1 characters as constants",
         "psql": "8.3",
         "description": "test inline UTF8 constants",
         "sql": "SELECT 'Οὐχὶ (greek)' FROM {varchar}"