diff --git a/paimon-core/src/main/java/org/apache/paimon/casting/TimestampToNumericPrimitiveCastRule.java b/paimon-core/src/main/java/org/apache/paimon/casting/TimestampToNumericPrimitiveCastRule.java
index c772366115cf..f7cd21babd20 100644
--- a/paimon-core/src/main/java/org/apache/paimon/casting/TimestampToNumericPrimitiveCastRule.java
+++ b/paimon-core/src/main/java/org/apache/paimon/casting/TimestampToNumericPrimitiveCastRule.java
@@ -38,22 +38,34 @@ private TimestampToNumericPrimitiveCastRule() {
                 CastRulePredicate.builder()
                         .input(DataTypeRoot.TIMESTAMP_WITHOUT_TIME_ZONE)
                         .input(DataTypeRoot.TIMESTAMP_WITH_LOCAL_TIME_ZONE)
-                        .target(DataTypeFamily.NUMERIC)
+                        .target(DataTypeRoot.BIGINT)
+                        .target(DataTypeRoot.INTEGER)
                         .build());
     }
 
     @Override
     public CastExecutor<Timestamp, Number> create(DataType inputType, DataType targetType) {
-        if (targetType.is(DataTypeRoot.BIGINT)) {
-            if (inputType.is(DataTypeRoot.TIMESTAMP_WITHOUT_TIME_ZONE)) {
+        if (inputType.is(DataTypeRoot.TIMESTAMP_WITHOUT_TIME_ZONE)) {
+            if (targetType.is(DataTypeRoot.BIGINT)) {
                 return value -> DateTimeUtils.unixTimestamp(value.getMillisecond());
-            } else if (inputType.is(DataTypeRoot.TIMESTAMP_WITH_LOCAL_TIME_ZONE)) {
+            } else if (targetType.is(DataTypeRoot.INTEGER)) {
+                return value -> (int) DateTimeUtils.unixTimestamp(value.getMillisecond());
+            }
+        } else if (inputType.is(DataTypeRoot.TIMESTAMP_WITH_LOCAL_TIME_ZONE)) {
+            if (targetType.is(DataTypeRoot.BIGINT)) {
                 return value ->
                         DateTimeUtils.unixTimestamp(
                                 Timestamp.fromLocalDateTime(value.toLocalDateTime())
                                         .getMillisecond());
+            } else if (targetType.is(DataTypeRoot.INTEGER)) {
+                return value ->
+                        (int)
+                                DateTimeUtils.unixTimestamp(
+                                        Timestamp.fromLocalDateTime(value.toLocalDateTime())
+                                                .getMillisecond());
             }
         }
+
         return null;
     }
 }
diff --git a/paimon-core/src/test/java/org/apache/paimon/casting/CastExecutorTest.java b/paimon-core/src/test/java/org/apache/paimon/casting/CastExecutorTest.java
index 6d594d9752fb..5c41f2b5fcef 100644
--- a/paimon-core/src/test/java/org/apache/paimon/casting/CastExecutorTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/casting/CastExecutorTest.java
@@ -296,7 +296,7 @@ public void testTimestampToNumeric() {
                         DateTimeUtils.toLocalDateTime(mills, TimeZone.getDefault().toZoneId()));
         long millisecond1 = timestamp2.getMillisecond();
 
-        // cast from TimestampType to BigIntType
+        // cast from TimestampType to BigIntType or IntType
         compareCastResult(
                 CastExecutors.resolve(new TimestampType(3), new BigIntType(false)),
                 timestamp1,
@@ -307,7 +307,17 @@ public void testTimestampToNumeric() {
                 timestamp2,
                 DateTimeUtils.unixTimestamp(millisecond1));
 
-        // cast from BigIntType to TimestampType
+        compareCastResult(
+                CastExecutors.resolve(new TimestampType(3), new IntType(false)),
+                timestamp1,
+                (int) DateTimeUtils.unixTimestamp(millisecond));
+
+        compareCastResult(
+                CastExecutors.resolve(new LocalZonedTimestampType(3), new IntType(false)),
+                timestamp2,
+                (int) DateTimeUtils.unixTimestamp(millisecond1));
+
+        // cast from BigIntType or IntType to TimestampType
         compareCastResult(
                 CastExecutors.resolve(new BigIntType(false), new TimestampType(3)),
                 DateTimeUtils.unixTimestamp(millisecond),
@@ -317,6 +327,16 @@ public void testTimestampToNumeric() {
                 CastExecutors.resolve(new BigIntType(false), new LocalZonedTimestampType(3)),
                 DateTimeUtils.unixTimestamp(millisecond),
                 timestamp2);
+
+        compareCastResult(
+                CastExecutors.resolve(new IntType(false), new TimestampType(3)),
+                (int) DateTimeUtils.unixTimestamp(millisecond),
+                timestamp1);
+
+        compareCastResult(
+                CastExecutors.resolve(new IntType(false), new LocalZonedTimestampType(3)),
+                (int) DateTimeUtils.unixTimestamp(millisecond),
+                timestamp2);
     }
 
     @Test