[FLINK-1997][table] Add <> and != for un-equal in expressions

flink-staging/flink-table/src/main/scala/org/apache/flink/api/table/parser/ExpressionParser.scala

@@ -139,15 +139,11 @@ object ExpressionParser extends JavaTokenParsers with PackratParsers {
 
   // Comparison
 
-  lazy val equalTo: PackratParser[Expression] = term ~ "===" ~ term ^^ {
+  lazy val equalTo: PackratParser[Expression] = term ~ ("===" | "=") ~ term ^^ {
     case l ~ _ ~ r => EqualTo(l, r)
   }
 
-  lazy val equalToAlt: PackratParser[Expression] = term ~ "=" ~ term ^^ {
-    case l ~ _ ~ r => EqualTo(l, r)
-  }
-
-  lazy val notEqualTo: PackratParser[Expression] = term ~ "!==" ~ term ^^ {
+  lazy val notEqualTo: PackratParser[Expression] = term ~ ("!==" | "!=" | "<>") ~ term ^^ {
     case l ~ _ ~ r => NotEqualTo(l, r)
   }
 
@@ -168,7 +164,7 @@ object ExpressionParser extends JavaTokenParsers with PackratParsers {
   }
 
   lazy val comparison: PackratParser[Expression] =
-      equalTo | equalToAlt | notEqualTo |
+      equalTo | notEqualTo |
       greaterThan | greaterThanOrEqual |
       lessThan | lessThanOrEqual | term
 

flink-staging/flink-table/src/test/java/org/apache/flink/api/java/table/test/FilterITCase.java

@@ -118,7 +118,30 @@ public class FilterITCase extends MultipleProgramsTestBase {
 				tableEnv.toTable(input, "a, b, c");
 
 		Table result = table
-				.filter(" a % 2 === 0 ");
+				.filter(" a % 2 = 0 ");
+
+		DataSet<Row> ds = tableEnv.toSet(result, Row.class);
+		ds.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
+
+		env.execute();
+
+		expected = "2,2,Hello\n" + "4,3,Hello world, how are you?\n" + "6,3,Luke Skywalker\n" + "8,4," +
+				"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 testNotEquals() throws Exception {
+		ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+		TableEnvironment tableEnv = new TableEnvironment();
+
+		DataSet<Tuple3<Integer, Long, String>> input = CollectionDataSets.get3TupleDataSet(env);
+
+		Table table =
+				tableEnv.toTable(input, "a, b, c");
+
+		Table result = table
+				.filter("!( a % 2 <> 0 ) ");
 
 		DataSet<Row> ds = tableEnv.toSet(result, Row.class);
 		ds.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);