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 1fb872f6835e1..7b8052502929d 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
@@ -144,7 +144,7 @@ class ColumnStats {
 
         GenericRecord genericRecord = (GenericRecord) record;
 
-        final Object fieldVal = convertValueForSpecificDataTypes(field.schema(), genericRecord.get(field.name()), true);
+        final Object fieldVal = convertValueForSpecificDataTypes(field.schema(), genericRecord.get(field.name()), false);
         final Schema fieldSchema = getNestedFieldSchemaFromWriteSchema(genericRecord.getSchema(), field.name());
 
         if (fieldVal != null && canCompare(fieldSchema)) {
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/stats/ColumnStatsIndices.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/stats/ColumnStatsIndices.java
index 75e10341db79a..428661844eac4 100644
--- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/stats/ColumnStatsIndices.java
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/stats/ColumnStatsIndices.java
@@ -42,8 +42,6 @@
 import org.apache.flink.table.types.DataType;
 import org.apache.flink.table.types.logical.LogicalType;
 import org.apache.flink.table.types.logical.RowType;
-import org.apache.flink.table.types.logical.TimeType;
-import org.apache.flink.table.types.logical.TimestampType;
 
 import java.io.IOException;
 import java.util.Arrays;
@@ -273,29 +271,6 @@ private static Object doUnpack(
       Object rawVal,
       LogicalType logicalType,
       Map<LogicalType, AvroToRowDataConverters.AvroToRowDataConverter> converters) {
-    // fix time unit
-    switch (logicalType.getTypeRoot()) {
-      case TIME_WITHOUT_TIME_ZONE:
-        TimeType timeType = (TimeType) logicalType;
-        if (timeType.getPrecision() == 3) {
-          // the precision in HoodieMetadata is 6
-          rawVal = ((Long) rawVal) / 1000;
-        } else if (timeType.getPrecision() == 9) {
-          rawVal = ((Long) rawVal) * 1000;
-        }
-        break;
-      case TIMESTAMP_WITHOUT_TIME_ZONE:
-        TimestampType timestampType = (TimestampType) logicalType;
-        if (timestampType.getPrecision() == 3) {
-          // the precision in HoodieMetadata is 6
-          rawVal = ((Long) rawVal) / 1000;
-        } else if (timestampType.getPrecision() == 9) {
-          rawVal = ((Long) rawVal) * 1000;
-        }
-        break;
-      default:
-        // no operation
-    }
     AvroToRowDataConverters.AvroToRowDataConverter converter =
         converters.computeIfAbsent(logicalType, k -> AvroToRowDataConverters.createConverter(logicalType));
     return converter.convert(rawVal);
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/stats/ExpressionEvaluator.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/stats/ExpressionEvaluator.java
index 6808e3f651e12..08ded144e0a4b 100644
--- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/stats/ExpressionEvaluator.java
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/stats/ExpressionEvaluator.java
@@ -32,6 +32,7 @@
 import org.apache.flink.table.types.logical.DecimalType;
 import org.apache.flink.table.types.logical.LogicalType;
 import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.table.types.logical.TimestampType;
 
 import javax.validation.constraints.NotNull;
 
@@ -523,6 +524,8 @@ private static Object getValAsJavaObj(RowData indexRow, int pos, LogicalType col
       //       manually encoding corresponding values as int and long w/in the Column Stats Index and
       //       here we have to decode those back into corresponding logical representation.
       case TIMESTAMP_WITHOUT_TIME_ZONE:
+        TimestampType tsType = (TimestampType) colType;
+        return indexRow.getTimestamp(pos, tsType.getPrecision()).getMillisecond();
       case TIME_WITHOUT_TIME_ZONE:
       case DATE:
         return indexRow.getLong(pos);