diff --git a/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java b/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java
index 2c5b8db0edc37..c7a0df5d6ad33 100644
--- a/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java
+++ b/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java
@@ -1323,6 +1323,8 @@ private static Comparable<?> coerceToComparable(Schema schema, Object val) {
         return (Long) val;
 
       case STRING:
+        // unpack the avro Utf8 if possible
+        return val.toString();
       case FLOAT:
       case DOUBLE:
       case BOOLEAN:
diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestHoodieDataSource.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestHoodieDataSource.java
index c40831639b67f..e40c40c996092 100644
--- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestHoodieDataSource.java
+++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestHoodieDataSource.java
@@ -1297,14 +1297,16 @@ void testBuiltinFunctionWithCatalog(String operation) {
     assertRowsEquals(partitionResult, "[+I[1, 2022-02-02]]");
   }
 
-  @Test
-  void testWriteAndReadWithDataSkipping() {
+  @ParameterizedTest
+  @EnumSource(value = HoodieTableType.class)
+  void testWriteAndReadWithDataSkipping(HoodieTableType tableType) {
     TableEnvironment tableEnv = batchTableEnv;
     String hoodieTableDDL = sql("t1")
         .option(FlinkOptions.PATH, tempFile.getAbsolutePath())
         .option(FlinkOptions.METADATA_ENABLED, true)
         .option("hoodie.metadata.index.column.stats.enable", true)
         .option(FlinkOptions.READ_DATA_SKIPPING_ENABLED, true)
+        .option(FlinkOptions.TABLE_TYPE,tableType)
         .end();
     tableEnv.executeSql(hoodieTableDDL);