diff --git a/flink-staging/flink-table/src/main/scala/org/apache/flink/api/table/codegen/ExpressionCodeGenerator.scala b/flink-staging/flink-table/src/main/scala/org/apache/flink/api/table/codegen/ExpressionCodeGenerator.scala
index 10f58590ee12ad32e8cd80dd3108454dce0326dc..a03ba618870dfe8d5ef2e02b1b44f08f0ba32597 100644
--- a/flink-staging/flink-table/src/main/scala/org/apache/flink/api/table/codegen/ExpressionCodeGenerator.scala
+++ b/flink-staging/flink-table/src/main/scala/org/apache/flink/api/table/codegen/ExpressionCodeGenerator.scala
@@ -409,7 +409,7 @@ abstract class ExpressionCodeGenerator[R](
           """.stripMargin
         } else {
           s"""
-            |$resultTerm = ${childGen.resultTerm}.getTime();
+            |$resultTpe $resultTerm = ${childGen.resultTerm}.getTime();
           """.stripMargin
         }
         childGen.code + castCode
diff --git a/flink-staging/flink-table/src/test/java/org/apache/flink/api/java/table/test/CastingITCase.java b/flink-staging/flink-table/src/test/java/org/apache/flink/api/java/table/test/CastingITCase.java
index 6a83d175c01eb774e58f25942085d142a8542581..7e9e3dcadef1351cc16d1ed3c54aa0cd17433f49 100644
--- a/flink-staging/flink-table/src/test/java/org/apache/flink/api/java/table/test/CastingITCase.java
+++ b/flink-staging/flink-table/src/test/java/org/apache/flink/api/java/table/test/CastingITCase.java
@@ -18,6 +18,7 @@
 
 package org.apache.flink.api.java.table.test;
 
+import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.api.java.tuple.Tuple3;
 import org.apache.flink.api.java.tuple.Tuple4;
 import org.apache.flink.api.table.Table;
@@ -145,5 +146,26 @@ public class CastingITCase extends MultipleProgramsTestBase {
 				"1970-01-17 17:47:53.775\n";
 		compareResultAsText(results, expected);
 	}
+
+	@Test
+	public void testCastDateToStringAndLong() throws Exception {
+		ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+		TableEnvironment tableEnv = new TableEnvironment();
+
+		DataSource<Tuple2<String, String>> input =
+			env.fromElements(new Tuple2<>("2011-05-03 15:51:36.000", "1304437896000"));
+
+		Table table =
+			tableEnv.fromDataSet(input);
+
+		Table result = table
+			.select("f0.cast(DATE) AS f0, f1.cast(DATE) AS f1")
+			.select("f0.cast(STRING), f0.cast(LONG), f1.cast(STRING), f1.cast(LONG)");
+
+		DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+		List<Row> results = ds.collect();
+		String expected = "2011-05-03 15:51:36.000,1304437896000,2011-05-03 15:51:36.000,1304437896000\n";
+		compareResultAsText(results, expected);
+	}
 }
 
diff --git a/flink-staging/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/CastingITCase.scala b/flink-staging/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/CastingITCase.scala
index 77fd7b78ecba2fda00c0a65f9292f3d855d21d75..c177184159ada655050c0ad86f639f1bb2f52cf7 100644
--- a/flink-staging/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/CastingITCase.scala
+++ b/flink-staging/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/CastingITCase.scala
@@ -105,4 +105,21 @@ class CastingITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mo
     TestBaseUtils.compareResultAsText(results.asJava, expected)
   }
 
+  @Test
+  def testCastDateToStringAndLong {
+    val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
+    val ds = env.fromElements(("2011-05-03 15:51:36.000", "1304437896000"))
+    val result = ds.toTable
+      .select('_1.cast(BasicTypeInfo.DATE_TYPE_INFO).as('f0),
+        '_2.cast(BasicTypeInfo.DATE_TYPE_INFO).as('f1))
+      .select('f0.cast(BasicTypeInfo.STRING_TYPE_INFO),
+        'f0.cast(BasicTypeInfo.LONG_TYPE_INFO),
+        'f1.cast(BasicTypeInfo.STRING_TYPE_INFO),
+        'f1.cast(BasicTypeInfo.LONG_TYPE_INFO))
+      .toDataSet[Row]
+      .collect
+    val expected = "2011-05-03 15:51:36.000,1304437896000," +
+      "2011-05-03 15:51:36.000,1304437896000\n"
+    TestBaseUtils.compareResultAsText(result.asJava, expected)
+  }
 }