diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala
index 8f0b824a2faa3..083a8bd0407a2 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala
@@ -1140,11 +1140,20 @@ object ScalarOperatorGens {
       }
 
     case (VARCHAR | CHAR, TIMESTAMP_WITH_LOCAL_TIME_ZONE) =>
-      generateUnaryOperatorIfNotNull(
-        ctx, targetType, operand, resultNullable = true) { operandTerm =>
+      generateCallWithStmtIfArgsNotNull(
+        ctx, targetType, Seq(operand), resultNullable = true) { operands =>
         val zone = ctx.addReusableSessionTimeZone()
         val method = qualifyMethod(BuiltInMethods.STRING_TO_TIMESTAMP_TIME_ZONE)
-        s"$TIMESTAMP_DATA.fromEpochMillis($method($operandTerm.toString(), $zone))"
+        val toTimestampResultName = newName("toTimestampResult")
+        // this method call might return null
+        val stmt = s"Long $toTimestampResultName = $method(${operands.head}.toString(), $zone);"
+        val result =
+          s"""
+             |($toTimestampResultName == null ?
+             |  null :
+             |  $TIMESTAMP_DATA.fromEpochMillis($toTimestampResultName))
+             |""".stripMargin
+        (stmt, result)
       }
 
     // String -> binary
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/expressions/ScalarOperatorsTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/expressions/ScalarOperatorsTest.scala
index 046293f83f08a..bb34a405287fe 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/expressions/ScalarOperatorsTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/expressions/ScalarOperatorsTest.scala
@@ -196,4 +196,55 @@ class ScalarOperatorsTest extends ScalarOperatorsTestBase {
     testSqlApi("'1996-11-10 12:34:57' = f22", "false")
     testSqlApi("cast(null as string) = f22", "null")
   }
+
+  @Test
+  def testTemporalTypeEqualsStringType(): Unit = {
+    testSqlApi("f15 = date_format(cast(f15 as timestamp), 'yyyy-MM-dd')", "true")
+    testSqlApi(
+      "f15 = date_format(cast(f15 as timestamp) + interval '1' day, 'yyyy-MM-dd')",
+      "false")
+    testSqlApi("f15 = uuid()", "null")
+    testSqlApi("date_format(cast(f15 as timestamp), 'yyyy-MM-dd') = f15", "true")
+    testSqlApi(
+      "date_format(cast(f15 as timestamp) + interval '1' day, 'yyyy-MM-dd') = f15",
+      "false")
+    testSqlApi("uuid() = f15", "null")
+
+    testSqlApi("f21 = date_format(cast(f21 as timestamp), 'HH:mm:ss')", "true")
+    testSqlApi(
+      "f21 = date_format(cast(f21 as timestamp) + interval '1' hour, 'HH:mm:ss')",
+      "false")
+    testSqlApi("f21 = uuid()", "null")
+    testSqlApi("date_format(cast(f21 as timestamp), 'HH:mm:ss') = f21", "true")
+    testSqlApi(
+      "date_format(cast(f21 as timestamp) + interval '1' hour, 'HH:mm:ss') = f21",
+      "false")
+    testSqlApi("uuid() = f21", "null")
+
+    testSqlApi("f22 = date_format(f22, 'yyyy-MM-dd HH:mm:ss')", "true")
+    testSqlApi(
+      "f22 = date_format(f22 + interval '1' second, 'yyyy-MM-dd HH:mm:ss')",
+      "false")
+    testSqlApi("f22 = uuid()", "null")
+    testSqlApi("date_format(f22, 'yyyy-MM-dd HH:mm:ss') = f22", "true")
+    testSqlApi(
+      "date_format(f22 + interval '1' second, 'yyyy-MM-dd HH:mm:ss') = f22",
+      "false")
+    testSqlApi("uuid() = f22", "null")
+
+    testSqlApi(
+      "cast(f22 as timestamp_ltz) = date_format(f22, 'yyyy-MM-dd HH:mm:ss')",
+      "true")
+    testSqlApi(
+      "cast(f22 as timestamp_ltz) = date_format(f22 + interval '1' second, 'yyyy-MM-dd HH:mm:ss')",
+      "false")
+    testSqlApi("cast(f22 as timestamp_ltz) = uuid()", "null")
+    testSqlApi(
+      "date_format(f22, 'yyyy-MM-dd HH:mm:ss') = cast(f22 as timestamp_ltz)",
+      "true")
+    testSqlApi(
+      "date_format(f22 + interval '1' second, 'yyyy-MM-dd HH:mm:ss') = cast(f22 as timestamp_ltz)",
+      "false")
+    testSqlApi("uuid() = cast(f22 as timestamp_ltz)", "null")
+  }
 }