diff --git a/flink-staging/flink-table/src/main/scala/org/apache/flink/api/java/table/JavaBatchTranslator.scala b/flink-staging/flink-table/src/main/scala/org/apache/flink/api/java/table/JavaBatchTranslator.scala
index 7653ba8681640..24e50efa93cee 100644
--- a/flink-staging/flink-table/src/main/scala/org/apache/flink/api/java/table/JavaBatchTranslator.scala
+++ b/flink-staging/flink-table/src/main/scala/org/apache/flink/api/java/table/JavaBatchTranslator.scala
@@ -251,7 +251,7 @@ class JavaBatchTranslator extends PlanTranslator {
         val translatedInput = translateInternal(input)
         val inType = translatedInput.getType.asInstanceOf[CompositeType[Row]]
         val filter = new ExpressionFilterFunction[Row](predicate, inType)
-        translatedInput.filter(filter)
+        translatedInput.filter(filter).name(predicate.toString)
     }
   }
 
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 fc0abe4cfa086..8cb678aff1c0c 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
@@ -317,7 +317,7 @@ abstract class ExpressionCodeGenerator[R](
 
       case EqualTo(left, right) =>
         generateIfNonNull(left, right, BOOLEAN_TYPE_INFO) {
-          (leftTerm, rightTerm) => s"$leftTerm == $rightTerm"
+          (leftTerm, rightTerm) => s"$leftTerm.equals($rightTerm)"
         }
 
       case NotEqualTo(left, right) =>
diff --git a/flink-staging/flink-table/src/test/java/org/apache/flink/api/java/table/test/FilterITCase.java b/flink-staging/flink-table/src/test/java/org/apache/flink/api/java/table/test/FilterITCase.java
index d40b6a03b9469..8b5fc43f13631 100644
--- a/flink-staging/flink-table/src/test/java/org/apache/flink/api/java/table/test/FilterITCase.java
+++ b/flink-staging/flink-table/src/test/java/org/apache/flink/api/java/table/test/FilterITCase.java
@@ -152,5 +152,24 @@ public void testNotEquals() throws Exception {
 				"Comment#2\n" + "10,4,Comment#4\n" + "12,5,Comment#6\n" + "14,5,Comment#8\n" + "16,6," +
 				"Comment#10\n" + "18,6,Comment#12\n" + "20,6,Comment#14\n";
 	}
+
+	@Test
+	public void testIntegerBiggerThan128() throws Exception {
+		ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+		TableEnvironment tableEnv = new TableEnvironment();
+
+		DataSet<Tuple3<Integer, Long, String>> input = env.fromElements(new Tuple3<Integer, Long, String>(300, 1L, "Hello"));
+
+		Table table = tableEnv.toTable(input, "a, b, c");
+
+		Table result = table.filter("a = 300 ");
+
+		DataSet<Row> ds = tableEnv.toSet(result, Row.class);
+		ds.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
+
+		env.execute();
+
+		expected = "300,1,Hello\n";
+	}
 }