diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/ExpressionPredicates.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/ExpressionPredicates.java
index 046e4b739ada..b77fc8478e18 100644
--- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/ExpressionPredicates.java
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/ExpressionPredicates.java
@@ -602,10 +602,10 @@ private static FilterPredicate toParquetPredicate(FunctionDefinition functionDef
       case TINYINT:
       case SMALLINT:
       case INTEGER:
+      case DATE:
       case TIME_WITHOUT_TIME_ZONE:
         return predicateSupportsLtGt(functionDefinition, intColumn(columnName), (Integer) literal);
       case BIGINT:
-      case DATE:
       case TIMESTAMP_WITHOUT_TIME_ZONE:
         return predicateSupportsLtGt(functionDefinition, longColumn(columnName), (Long) literal);
       case FLOAT:
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/ExpressionUtils.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/ExpressionUtils.java
index 78245fb80a0d..1783057beeb7 100644
--- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/ExpressionUtils.java
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/ExpressionUtils.java
@@ -160,7 +160,7 @@ public static Object getValueFromLiteral(ValueLiteralExpression expr) {
             .orElse(null);
       case DATE:
         return expr.getValueAs(LocalDate.class)
-            .map(LocalDate::toEpochDay)
+            .map(date -> (int) date.toEpochDay())
             .orElse(null);
       // NOTE: All integral types of size less than Int are encoded as Ints in MT
       case BOOLEAN:
@@ -212,7 +212,7 @@ public static Object getKeyFromLiteral(ValueLiteralExpression expr, boolean logi
       case TIMESTAMP_WITHOUT_TIME_ZONE:
         return logicalTimestamp ? new Timestamp((long) val) : val;
       case DATE:
-        return LocalDate.ofEpochDay((long) val);
+        return LocalDate.ofEpochDay((int) val);
       default:
         return val;
     }
diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/util/TestExpressionUtils.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/util/TestExpressionUtils.java
index c9eb5ac54959..64c205a8f716 100644
--- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/util/TestExpressionUtils.java
+++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/util/TestExpressionUtils.java
@@ -140,7 +140,7 @@ void getValueFromLiteralForNonNull() {
       if (dataList.get(i) instanceof LocalTime) {
         assertEquals(((LocalTime) dataList.get(i)).get(ChronoField.MILLI_OF_DAY), ExpressionUtils.getValueFromLiteral((ValueLiteralExpression) childExprs.get(1)));
       } else if (dataList.get(i) instanceof LocalDate) {
-        assertEquals(((LocalDate) dataList.get(i)).toEpochDay(), ExpressionUtils.getValueFromLiteral((ValueLiteralExpression) childExprs.get(1)));
+        assertEquals((int) ((LocalDate) dataList.get(i)).toEpochDay(), ExpressionUtils.getValueFromLiteral((ValueLiteralExpression) childExprs.get(1)));
       } else if (dataList.get(i) instanceof LocalDateTime) {
         assertEquals(((LocalDateTime) dataList.get(i)).toInstant(ZoneOffset.UTC).toEpochMilli(), ExpressionUtils.getValueFromLiteral((ValueLiteralExpression) childExprs.get(1)));
       } else {