[FLINK-35098][ORC] Fix incorrect results with literal first expressions

apache · May 14, 2024 · 4165bac · 4165bac
1 parent ddb5a53
commit 4165bac
Show file tree

Hide file tree

Showing 3 changed files with 65 additions and 12 deletions.
diff --git a/flink-formats/flink-orc/src/main/java/org/apache/flink/orc/OrcFilters.java b/flink-formats/flink-orc/src/main/java/org/apache/flink/orc/OrcFilters.java
@@ -83,28 +83,28 @@ public class OrcFilters {
                                             convertBinary(
                                                     call,
                                                     OrcFilters::convertGreaterThan,
-                                                    OrcFilters::convertLessThanEquals))
+                                                    OrcFilters::convertLessThan))
                             .put(
                                     BuiltInFunctionDefinitions.GREATER_THAN_OR_EQUAL,
                                     call ->
                                             convertBinary(
                                                     call,
                                                     OrcFilters::convertGreaterThanEquals,
-                                                    OrcFilters::convertLessThan))
+                                                    OrcFilters::convertLessThanEquals))
                             .put(
                                     BuiltInFunctionDefinitions.LESS_THAN,
                                     call ->
                                             convertBinary(
                                                     call,
                                                     OrcFilters::convertLessThan,
-                                                    OrcFilters::convertGreaterThanEquals))
+                                                    OrcFilters::convertGreaterThan))
                             .put(
                                     BuiltInFunctionDefinitions.LESS_THAN_OR_EQUAL,
                                     call ->
                                             convertBinary(
                                                     call,
                                                     OrcFilters::convertLessThanEquals,
-                                                    OrcFilters::convertGreaterThan))
+                                                    OrcFilters::convertGreaterThanEquals))
                             .build();
 
     private static boolean isRef(Expression expression) {

diff --git a/flink-formats/flink-orc/src/test/java/org/apache/flink/orc/OrcFileSystemFilterTest.java b/flink-formats/flink-orc/src/test/java/org/apache/flink/orc/OrcFileSystemFilterTest.java
@@ -25,21 +25,21 @@
 import org.apache.flink.table.expressions.ValueLiteralExpression;
 import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
 
-import org.apache.hadoop.hive.ql.io.sarg.PredicateLeaf;
 import org.junit.jupiter.api.Test;
 
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 
+import static org.apache.hadoop.hive.ql.io.sarg.PredicateLeaf.Type.LONG;
 import static org.assertj.core.api.Assertions.assertThat;
 
 /** Unit Tests for {@link OrcFileFormatFactory}. */
 class OrcFileSystemFilterTest {
 
     @Test
     @SuppressWarnings("unchecked")
-    public void testApplyPredicate() {
+    void testApplyPredicate() {
         List<ResolvedExpression> args = new ArrayList<>();
 
         // equal
@@ -53,8 +53,7 @@ public void testApplyPredicate() {
                 CallExpression.permanent(
                         BuiltInFunctionDefinitions.EQUALS, args, DataTypes.BOOLEAN());
         OrcFilters.Predicate predicate1 = OrcFilters.toOrcPredicate(equalExpression);
-        OrcFilters.Predicate predicate2 =
-                new OrcFilters.Equals("long1", PredicateLeaf.Type.LONG, 10);
+        OrcFilters.Predicate predicate2 = new OrcFilters.Equals("long1", LONG, 10);
         assertThat(predicate1).hasToString(predicate2.toString());
 
         // greater than
@@ -63,17 +62,15 @@ public void testApplyPredicate() {
                         BuiltInFunctionDefinitions.GREATER_THAN, args, DataTypes.BOOLEAN());
         OrcFilters.Predicate predicate3 = OrcFilters.toOrcPredicate(greaterExpression);
         OrcFilters.Predicate predicate4 =
-                new OrcFilters.Not(
-                        new OrcFilters.LessThanEquals("long1", PredicateLeaf.Type.LONG, 10));
+                new OrcFilters.Not(new OrcFilters.LessThanEquals("long1", LONG, 10));
         assertThat(predicate3).hasToString(predicate4.toString());
 
         // less than
         CallExpression lessExpression =
                 CallExpression.permanent(
                         BuiltInFunctionDefinitions.LESS_THAN, args, DataTypes.BOOLEAN());
         OrcFilters.Predicate predicate5 = OrcFilters.toOrcPredicate(lessExpression);
-        OrcFilters.Predicate predicate6 =
-                new OrcFilters.LessThan("long1", PredicateLeaf.Type.LONG, 10);
+        OrcFilters.Predicate predicate6 = new OrcFilters.LessThan("long1", LONG, 10);
         assertThat(predicate5).hasToString(predicate6.toString());
 
         // and
@@ -86,4 +83,49 @@ public void testApplyPredicate() {
         OrcFilters.Predicate predicate8 = new OrcFilters.And(predicate4, predicate6);
         assertThat(predicate7).hasToString(predicate8.toString());
     }
+
+    @Test
+    @SuppressWarnings("unchecked")
+    void testApplyPredicateReverse() {
+        List<ResolvedExpression> args = new ArrayList<>();
+
+        FieldReferenceExpression fieldReferenceExpression =
+                new FieldReferenceExpression("x", DataTypes.BIGINT(), 0, 0);
+        ValueLiteralExpression valueLiteralExpression = new ValueLiteralExpression(10);
+        args.add(valueLiteralExpression);
+        args.add(fieldReferenceExpression);
+
+        CallExpression expression;
+        OrcFilters.Predicate predicate;
+
+        // assert that 10 >= x expression is converted to x <= 10 ORC predicate
+        expression =
+                CallExpression.permanent(
+                        BuiltInFunctionDefinitions.GREATER_THAN_OR_EQUAL,
+                        args,
+                        DataTypes.BOOLEAN());
+        predicate = new OrcFilters.LessThanEquals("x", LONG, 10);
+        assertThat(OrcFilters.toOrcPredicate(expression)).hasToString(predicate.toString());
+
+        // assert that 10 > x expression is converted to x < 10 ORC predicate
+        expression =
+                CallExpression.permanent(
+                        BuiltInFunctionDefinitions.GREATER_THAN, args, DataTypes.BOOLEAN());
+        predicate = new OrcFilters.LessThan("x", LONG, 10);
+        assertThat(OrcFilters.toOrcPredicate(expression)).hasToString(predicate.toString());
+
+        // assert that 10 <= x expression is converted to NOT(x < 10) ORC predicate
+        expression =
+                CallExpression.permanent(
+                        BuiltInFunctionDefinitions.LESS_THAN_OR_EQUAL, args, DataTypes.BOOLEAN());
+        predicate = new OrcFilters.Not(new OrcFilters.LessThan("x", LONG, 10));
+        assertThat(OrcFilters.toOrcPredicate(expression)).hasToString(predicate.toString());
+
+        // assert that 10 < x expression is converted to NOT(x <= 10) ORC predicate
+        expression =
+                CallExpression.permanent(
+                        BuiltInFunctionDefinitions.LESS_THAN, args, DataTypes.BOOLEAN());
+        predicate = new OrcFilters.Not(new OrcFilters.LessThanEquals("x", LONG, 10));
+        assertThat(OrcFilters.toOrcPredicate(expression)).hasToString(predicate.toString());
+    }
 }
diff --git a/flink-formats/flink-orc/src/test/java/org/apache/flink/orc/OrcFileSystemITCase.java b/flink-formats/flink-orc/src/test/java/org/apache/flink/orc/OrcFileSystemITCase.java
@@ -208,6 +208,17 @@ void testOrcFilterPushDown() throws ExecutionException, InterruptedException {
                 Collections.singletonList(Row.of("x10", "10")));
     }
 
+    @TestTemplate
+    void testOrcFilterPushDownLiteralFirst() throws ExecutionException, InterruptedException {
+        super.tableEnv()
+                .executeSql("insert into orcLimitTable values('a', 10, 10), ('b', 11, 11)")
+                .await();
+        check("select y from orcLimitTable where 10 >= y", Collections.singletonList(Row.of(10)));
+        check("select y from orcLimitTable where 11 <= y", Collections.singletonList(Row.of(11)));
+        check("select y from orcLimitTable where 11 > y", Collections.singletonList(Row.of(10)));
+        check("select y from orcLimitTable where 10 < y", Collections.singletonList(Row.of(11)));
+    }
+
     @TestTemplate
     void testNestedTypes() throws Exception {
         String path = initNestedTypesFile(initNestedTypesData());