[impala] 02/05: IMPALA-11446: Push-down NOT_IN predicate to iceberg

[st...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

stigahuang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git

commit 614dc54dcc14687a11bb64495d4b6d4356ef55fc
Author: LPL <li...@sensorsdata.cn>
AuthorDate: Wed Jul 20 14:31:32 2022 +0800

    IMPALA-11446: Push-down NOT_IN predicate to iceberg
    
    Because the column value bounds of the Iceberg meta are not necessarily
    a min or max value, NOT_IN cannot be answered using them.
    NOT_IN(col, {X, ...}) with (X, Y) doesn't guarantee that X is a value
    in col. But it works when the push-down column is the partition column,
    it's still very helpful.
    
    Testing:
      - add e2e tests
    
    Change-Id: Ib8bdaf6f31a4438e11c4eb27485bb413fe6df9a3
    Reviewed-on: http://gerrit.cloudera.org:8080/18760
    Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
    Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
 .../org/apache/impala/planner/IcebergScanNode.java | 14 +++++++-----
 .../QueryTest/iceberg-in-predicate-push-down.test  | 26 ++++++++++++++++++++++
 2 files changed, 34 insertions(+), 6 deletions(-)

diff --git a/fe/src/main/java/org/apache/impala/planner/IcebergScanNode.java b/fe/src/main/java/org/apache/impala/planner/IcebergScanNode.java
index 547b02e0d..d065bc723 100644
--- a/fe/src/main/java/org/apache/impala/planner/IcebergScanNode.java
+++ b/fe/src/main/java/org/apache/impala/planner/IcebergScanNode.java
@@ -284,11 +284,6 @@ public class IcebergScanNode extends HdfsScanNode {
 
   private UnboundPredicate<Object> convertIcebergPredicate(Analyzer analyzer,
       InPredicate predicate) throws ImpalaException {
-    // TODO: convert NOT_IN predicate
-    if (predicate.isNotIn()) {
-      return null;
-    }
-
     // Do not convert if there is an implicit cast
     if (!(predicate.getChild(0) instanceof SlotRef)) {
       return null;
@@ -322,7 +317,14 @@ public class IcebergScanNode extends HdfsScanNode {
       values.add(value);
     }
 
-    return Expressions.in(col.getName(), values);
+    // According to the method:
+    // 'org.apache.iceberg.expressions.InclusiveMetricsEvaluator.MetricsEvalVisitor#notIn'
+    // Expressions.notIn only works when the push-down column is the partition column
+    if (predicate.isNotIn())
+      return Expressions.notIn(col.getName(), values);
+    else {
+      return Expressions.in(col.getName(), values);
+    }
   }
 
   private UnboundPredicate<Object> convertIcebergPredicate(IsNullPredicate predicate) {
diff --git a/testdata/workloads/functional-query/queries/QueryTest/iceberg-in-predicate-push-down.test b/testdata/workloads/functional-query/queries/QueryTest/iceberg-in-predicate-push-down.test
index 91144dd64..319165fb2 100644
--- a/testdata/workloads/functional-query/queries/QueryTest/iceberg-in-predicate-push-down.test
+++ b/testdata/workloads/functional-query/queries/QueryTest/iceberg-in-predicate-push-down.test
@@ -416,6 +416,32 @@ where
 aggregation(SUM, NumRowGroups): 0
 ====
 ---- QUERY
+# NOT_IN works when col_i is the partition column
+select
+  count(1)
+from
+  ice_pred_pd1
+where
+  col_i not in (0, 1);
+---- RESULTS
+3
+---- RUNTIME_PROFILE
+aggregation(SUM, NumRowGroups): 1
+====
+---- QUERY
+# NOT_IN does not work because col_dt is not the partition column
+select
+  count(1)
+from
+  ice_pred_pd1
+where
+  col_dt not in (DATE'1400-01-01', DATE'1500-01-01');
+---- RESULTS
+3
+---- RUNTIME_PROFILE
+aggregation(SUM, NumRowGroups): 3
+====
+---- QUERY
 # Test DECIMAL(9, 3)/DECIMAL(18, 3)/DECIMAL(38, 3) InPredicate push-down
 create table ice_pred_pd2 (
   p INT,