You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by mi...@apache.org on 2023/09/28 17:36:10 UTC
[impala] 02/03: IMPALA-12089: Be able to skip pushing down a subset of the predicates in Iceberg
This is an automated email from the ASF dual-hosted git repository.
michaelsmith pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 1f82106aff6fc2f0afb5a2c8ed754463955813a4
Author: Peter Rozsa <pr...@cloudera.com>
AuthorDate: Wed May 24 15:11:58 2023 +0200
IMPALA-12089: Be able to skip pushing down a subset of the predicates in Iceberg
This change adds a predicate filtering mechanism at planning time that
locates Impala's predicates in the residual expressions from Iceberg
planning. By locating all residual expressions, the remainder
expression set can be calculated.
The current implementation is an all-or-nothing filter, if 'planFiles()'
(Iceberg API) returns no residual expression, then all Impala
predicates can be skipped, if there's any residual expression, every
Impala predicate is pushed down to the Impala scanner.
Residual expressions are the remaining filter expressions after the
pushdown of predicates into the Iceberg table scan. By locating the
remainder expression, we can reduce the number of predicates that will
be pushed down to the Impala scanner.
After this change, the Iceberg residual expression handling is improved
by locating the simple conjuncts in the residual expression and mapping
back them to Impala conjuncts. For example, if the list of Impala
conjuncts consists of two predicates 'col_i != 100' and 'col_s = "a"'
and 'col_i' happens to be a partition column in the Iceberg table
definition and Iceberg table scan can eliminate the expression, the
residual expression will be 'col_s = "a"'. This expression can be mapped
back as an Impala predicate, and any other expression can be removed
from the effective Impala conjunct list, and pushed down to the scanner,
skipping the unnecessary filtering of 'col_i'.
If there's no residual expression, the behavior is the same as before,
all predicate pushdown is skipped.
If Impala is unable to match all residual expression to Impala conjuncts
then all the conjunct are pushed dow to Impala scanner.
This change offers the advantage of not pushing down already evaluated
filters to the Impala scanner nodes, resulting in enhanced scanning
performance. Additionally, if the filter expression affects columns that
are unnecessary for the final result and can be filtered out during
Iceberg's table scan, it leads to a reduced row size, thereby optimizing
data retrieval and improving overall query efficiency.
This solution is limited to cases where Impala's expression list
contains only conjuncts, compound expressions are not supported, because
partial elimination of compounds would involve expression rewrites in
the Impala expression.
A new query option is added: iceberg_predicate_pushdown_subsetting. The
query option's default value is true. It can be turned off by setting it
to false.
Performance of the predicate location is measured on two edge cases:
- 1000 expression, 999 skipped: on avreage 2 ms
- 1000 expression, 1 skipped: on average 25 ms
Tests:
- planner test cases added for disabled mode
- existing planner test cases adjusted
- core tests passed
Change-Id: I597f69ad03ecaf9e304613ef934654e3d9614ae8
Reviewed-on: http://gerrit.cloudera.org:8080/20133
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
be/src/service/query-options.cc | 3 +
be/src/service/query-options.h | 5 +-
common/thrift/ImpalaService.thrift | 4 +
common/thrift/Query.thrift | 3 +
.../analysis/IcebergExpressionCollector.java | 84 ++++++++++++++++
.../apache/impala/planner/IcebergScanPlanner.java | 88 +++++++++++------
.../org/apache/impala/planner/PlannerTest.java | 12 +++
.../iceberg-predicates-disabled-subsetting.test | 32 ++++++
.../queries/PlannerTest/iceberg-predicates.test | 108 +++++++++++++++++++--
.../PlannerTest/iceberg-v2-tables-hash-join.test | 48 ++++++---
.../queries/PlannerTest/iceberg-v2-tables.test | 48 ++++++---
.../queries/PlannerTest/tablesample.test | 7 +-
12 files changed, 370 insertions(+), 72 deletions(-)
diff --git a/be/src/service/query-options.cc b/be/src/service/query-options.cc
index 2db91041d..901acddbb 100644
--- a/be/src/service/query-options.cc
+++ b/be/src/service/query-options.cc
@@ -1144,6 +1144,9 @@ Status impala::SetQueryOption(const string& key, const string& value,
MemSpec mem_spec_val{};
RETURN_IF_ERROR(QueryOptionParser::Parse<MemSpec>(option, value, &mem_spec_val));
query_options->__set_mem_limit_coordinators(mem_spec_val.value);
+ }
+ case TImpalaQueryOptions::ICEBERG_PREDICATE_PUSHDOWN_SUBSETTING: {
+ query_options->__set_iceberg_predicate_pushdown_subsetting(IsTrue(value));
break;
}
default:
diff --git a/be/src/service/query-options.h b/be/src/service/query-options.h
index 74de1b972..654d5199e 100644
--- a/be/src/service/query-options.h
+++ b/be/src/service/query-options.h
@@ -50,7 +50,7 @@ typedef std::unordered_map<string, beeswax::TQueryOptionLevel::type>
// time we add or remove a query option to/from the enum TImpalaQueryOptions.
#define QUERY_OPTS_TABLE \
DCHECK_EQ(_TImpalaQueryOptions_VALUES_TO_NAMES.size(), \
- TImpalaQueryOptions::MEM_LIMIT_COORDINATORS + 1); \
+ TImpalaQueryOptions::ICEBERG_PREDICATE_PUSHDOWN_SUBSETTING + 1); \
REMOVED_QUERY_OPT_FN(abort_on_default_limit_exceeded, ABORT_ON_DEFAULT_LIMIT_EXCEEDED) \
QUERY_OPT_FN(abort_on_error, ABORT_ON_ERROR, TQueryOptionLevel::REGULAR) \
REMOVED_QUERY_OPT_FN(allow_unsupported_formats, ALLOW_UNSUPPORTED_FORMATS) \
@@ -306,7 +306,8 @@ typedef std::unordered_map<string, beeswax::TQueryOptionLevel::type>
TQueryOptionLevel::ADVANCED) \
QUERY_OPT_FN(mem_limit_coordinators, MEM_LIMIT_COORDINATORS, \
TQueryOptionLevel::ADVANCED) \
- ;
+ QUERY_OPT_FN(iceberg_predicate_pushdown_subsetting, \
+ ICEBERG_PREDICATE_PUSHDOWN_SUBSETTING, TQueryOptionLevel::DEVELOPMENT);
/// Enforce practical limits on some query options to avoid undesired query state.
static const int64_t SPILLABLE_BUFFER_LIMIT = 1LL << 40; // 1 TB
diff --git a/common/thrift/ImpalaService.thrift b/common/thrift/ImpalaService.thrift
index 0ee327216..660501513 100644
--- a/common/thrift/ImpalaService.thrift
+++ b/common/thrift/ImpalaService.thrift
@@ -841,6 +841,10 @@ enum TImpalaQueryOptions {
// a) an int (= number of bytes);
// b) a float followed by "M" (MB) or "G" (GB)
MEM_LIMIT_COORDINATORS = 164
+
+ // Enables predicate subsetting for Iceberg plan nodes. If enabled, expressions
+ // evaluated by Iceberg are not pushed down the scanner node.
+ ICEBERG_PREDICATE_PUSHDOWN_SUBSETTING = 165;
}
// The summary of a DML statement.
diff --git a/common/thrift/Query.thrift b/common/thrift/Query.thrift
index 033e756ed..edd827765 100644
--- a/common/thrift/Query.thrift
+++ b/common/thrift/Query.thrift
@@ -660,6 +660,9 @@ struct TQueryOptions {
// See comment in ImpalaService.thrift
165: optional i64 mem_limit_coordinators = 0;
+
+ // See comment in ImpalaService.thrift
+ 166: optional bool iceberg_predicate_pushdown_subsetting = true;
}
// Impala currently has three types of sessions: Beeswax, HiveServer2 and external
diff --git a/fe/src/main/java/org/apache/impala/analysis/IcebergExpressionCollector.java b/fe/src/main/java/org/apache/impala/analysis/IcebergExpressionCollector.java
new file mode 100644
index 000000000..742388a24
--- /dev/null
+++ b/fe/src/main/java/org/apache/impala/analysis/IcebergExpressionCollector.java
@@ -0,0 +1,84 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.impala.analysis;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import org.apache.iceberg.expressions.BoundAggregate;
+import org.apache.iceberg.expressions.BoundPredicate;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.ExpressionVisitors.ExpressionVisitor;
+import org.apache.iceberg.expressions.UnboundAggregate;
+import org.apache.iceberg.expressions.UnboundPredicate;
+
+/**
+ * Visitor implementation to traverse Iceberg expression tree and locate predicates.
+ */
+public class IcebergExpressionCollector extends ExpressionVisitor<List<Expression>> {
+ @Override
+ public List<Expression> alwaysTrue() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public List<Expression> alwaysFalse() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public List<Expression> not(List<Expression> result) {
+ return result;
+ }
+
+ @Override
+ public List<Expression> and(List<Expression> leftResult, List<Expression> rightResult) {
+ leftResult.addAll(rightResult);
+ return leftResult;
+ }
+
+ @Override
+ public List<Expression> or(List<Expression> leftResult, List<Expression> rightResult) {
+ leftResult.addAll(rightResult);
+ return leftResult;
+ }
+
+ @Override
+ public <T> List<Expression> predicate(BoundPredicate<T> pred) {
+ List<Expression> expressions = new ArrayList<>();
+ expressions.add(pred);
+ return expressions;
+ }
+
+ @Override
+ public <T> List<Expression> predicate(UnboundPredicate<T> pred) {
+ List<Expression> expressions = new ArrayList<>();
+ expressions.add(pred);
+ return expressions;
+ }
+
+ @Override
+ public <T, C> List<Expression> aggregate(BoundAggregate<T, C> agg) {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public <T> List<Expression> aggregate(UnboundAggregate<T> agg) {
+ return Collections.emptyList();
+ }
+}
diff --git a/fe/src/main/java/org/apache/impala/planner/IcebergScanPlanner.java b/fe/src/main/java/org/apache/impala/planner/IcebergScanPlanner.java
index d336f446c..0d4d37ca2 100644
--- a/fe/src/main/java/org/apache/impala/planner/IcebergScanPlanner.java
+++ b/fe/src/main/java/org/apache/impala/planner/IcebergScanPlanner.java
@@ -22,11 +22,14 @@ import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.TreeSet;
import java.util.function.Predicate;
import java.util.stream.Collectors;
@@ -37,6 +40,8 @@ import org.apache.iceberg.FileContent;
import org.apache.iceberg.FileScanTask;
import org.apache.iceberg.Schema;
import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.ExpressionUtil;
+import org.apache.iceberg.expressions.ExpressionVisitors;
import org.apache.iceberg.expressions.Expressions;
import org.apache.iceberg.expressions.Expression.Operation;
import org.apache.iceberg.expressions.True;
@@ -47,6 +52,7 @@ import org.apache.impala.analysis.Analyzer;
import org.apache.impala.analysis.BinaryPredicate;
import org.apache.impala.analysis.CompoundPredicate;
import org.apache.impala.analysis.Expr;
+import org.apache.impala.analysis.IcebergExpressionCollector;
import org.apache.impala.analysis.InPredicate;
import org.apache.impala.analysis.IsNullPredicate;
import org.apache.impala.analysis.JoinOperator;
@@ -80,7 +86,6 @@ import org.apache.impala.common.ImpalaException;
import org.apache.impala.common.ImpalaRuntimeException;
import org.apache.impala.common.InternalException;
import org.apache.impala.common.Pair;
-import org.apache.impala.planner.IcebergDeleteNode;
import org.apache.impala.planner.JoinNode.DistributionMode;
import org.apache.impala.thrift.TColumnStats;
import org.apache.impala.thrift.TIcebergPartitionTransformType;
@@ -101,29 +106,31 @@ import org.slf4j.LoggerFactory;
* class deals with such complexities.
*/
public class IcebergScanPlanner {
- private final static Logger LOG = LoggerFactory.getLogger(IcebergScanPlanner.class);
+ private static final Logger LOG = LoggerFactory.getLogger(IcebergScanPlanner.class);
private Analyzer analyzer_;
private PlannerContext ctx_;
private TableRef tblRef_;
private List<Expr> conjuncts_;
private MultiAggregateInfo aggInfo_;
-
- // Iceberg compatible expressions that are pushed down to Iceberg for query planning.
- private final List<Expression> icebergPredicates_ = new ArrayList<>();
- // The Impala representation of the expressions in 'icebergPredicates_'
- private final List<Expr> impalaIcebergPredicates_ = new ArrayList<>();
- // Indicates whether we have to push down 'impalaIcebergPredicates' to Impala's scan
- // node or has Iceberg already done the partition pruning and no further rows could be
- // skipped using these filters.
- private boolean canSkipPushingDownIcebergPredicates_ = false;
-
+ // Mapping for translated Impala expressions
+ private final Map<Expression, Expr> impalaIcebergPredicateMapping_ =
+ new LinkedHashMap<>();
+ // Residual expressions after Iceberg planning
+ private final Set<Expression> residualExpressions_ =
+ new TreeSet<>(Comparator.comparing(ExpressionUtil::toSanitizedString));
+ // Expressions filtered by Iceberg's planFiles, subset of 'translatedExpressions_''s
+ // values
+ private final Set<Expr> skippedExpressions_ =
+ new TreeSet<>(Comparator.comparingInt(System::identityHashCode));
+ // Impala expressions that can't be translated into Iceberg expressions
+ private final List<Expr> untranslatedExpressions_ = new ArrayList<>();
+ // Conjuncts on columns not involved in IDENTITY-partitioning.
+ private List<Expr> nonIdentityConjuncts_ = new ArrayList<>();
private List<FileDescriptor> dataFilesWithoutDeletes_ = new ArrayList<>();
private List<FileDescriptor> dataFilesWithDeletes_ = new ArrayList<>();
private Set<FileDescriptor> deleteFiles_ = new HashSet<>();
- // Conjuncts on columns not involved in IDENTITY-partitioning.
- private List<Expr> nonIdentityConjuncts_ = new ArrayList<>();
// Statistics about the data and delete files. Useful for memory estimates of the
// ANTI JOIN
@@ -163,9 +170,8 @@ public class IcebergScanPlanner {
* - no time travel
*/
private boolean needIcebergForPlanning() {
- return
- !icebergPredicates_.isEmpty() ||
- tblRef_.getTimeTravelSpec() != null;
+ return !impalaIcebergPredicateMapping_.isEmpty()
+ || tblRef_.getTimeTravelSpec() != null;
}
private void setFileDescriptorsBasedOnFileStore() throws ImpalaException {
@@ -366,14 +372,14 @@ public class IcebergScanPlanner {
private void filterFileDescriptors() throws ImpalaException {
TimeTravelSpec timeTravelSpec = tblRef_.getTimeTravelSpec();
- canSkipPushingDownIcebergPredicates_ = true;
- try (CloseableIterable<FileScanTask> fileScanTasks = IcebergUtil.planFiles(
- getIceTable(), new ArrayList<Expression>(icebergPredicates_), timeTravelSpec)) {
+ try (CloseableIterable<FileScanTask> fileScanTasks =
+ IcebergUtil.planFiles(getIceTable(),
+ new ArrayList<>(impalaIcebergPredicateMapping_.keySet()), timeTravelSpec)) {
long dataFilesCacheMisses = 0;
for (FileScanTask fileScanTask : fileScanTasks) {
Expression residualExpr = fileScanTask.residual();
if (residualExpr != null && !(residualExpr instanceof True)) {
- canSkipPushingDownIcebergPredicates_ = false;
+ residualExpressions_.add(residualExpr);
}
Pair<FileDescriptor, Boolean> fileDesc = getFileDescriptor(fileScanTask.file());
if (!fileDesc.second) ++dataFilesCacheMisses;
@@ -409,14 +415,42 @@ public class IcebergScanPlanner {
}
private void filterConjuncts() {
- if (canSkipPushingDownIcebergPredicates_) {
- conjuncts_.removeAll(impalaIcebergPredicates_);
+ if (residualExpressions_.isEmpty()) {
+ conjuncts_.removeAll(impalaIcebergPredicateMapping_.values());
+ return;
}
+ if (!analyzer_.getQueryOptions().iceberg_predicate_pushdown_subsetting) return;
+ trySubsettingPredicatesBeingPushedDown();
+ }
+
+ private boolean trySubsettingPredicatesBeingPushedDown() {
+ long startTime = System.currentTimeMillis();
+ List<Expr> expressionsToRetain = new ArrayList<>(untranslatedExpressions_);
+ for (Expression expression : residualExpressions_) {
+ List<Expression> locatedExpressions = ExpressionVisitors.visit(expression,
+ new IcebergExpressionCollector());
+ for (Expression located : locatedExpressions) {
+ Expr expr = impalaIcebergPredicateMapping_.get(located);
+ /* If we fail to locate any of the Iceberg residual expressions then we skip
+ filtering the predicates to be pushed down to Impala scanner.*/
+ if (expr == null) return false;
+ expressionsToRetain.add(expr);
+ }
+ }
+ skippedExpressions_.addAll(
+ conjuncts_.stream().filter(expr -> !expressionsToRetain.contains(expr)).collect(
+ Collectors.toSet()));
+ conjuncts_ = expressionsToRetain;
+ LOG.debug("Iceberg predicate pushdown subsetting took {} ms",
+ (System.currentTimeMillis() - startTime));
+ return true;
}
private List<Expr> getSkippedConjuncts() {
- if (!canSkipPushingDownIcebergPredicates_) return Collections.emptyList();
- return impalaIcebergPredicates_;
+ if (!residualExpressions_.isEmpty()) {
+ return new ArrayList<>(skippedExpressions_);
+ }
+ return new ArrayList<>(impalaIcebergPredicateMapping_.values());
}
private void updateDeleteStatistics() {
@@ -628,11 +662,11 @@ public class IcebergScanPlanner {
throws ImpalaException {
Expression predicate = convertIcebergPredicate(expr);
if (predicate != null) {
- icebergPredicates_.add(predicate);
- impalaIcebergPredicates_.add(expr);
+ impalaIcebergPredicateMapping_.put(predicate, expr);
LOG.debug("Push down the predicate: " + predicate + " to iceberg");
return true;
}
+ untranslatedExpressions_.add(expr);
return false;
}
diff --git a/fe/src/test/java/org/apache/impala/planner/PlannerTest.java b/fe/src/test/java/org/apache/impala/planner/PlannerTest.java
index 497dcb8a7..54c753eca 100644
--- a/fe/src/test/java/org/apache/impala/planner/PlannerTest.java
+++ b/fe/src/test/java/org/apache/impala/planner/PlannerTest.java
@@ -1274,6 +1274,18 @@ public class PlannerTest extends PlannerTestBase {
ImmutableSet.of(PlannerTestOption.VALIDATE_CARDINALITY));
}
+ /**
+ * Checks exercising predicate pushdown with Iceberg tables, without predicate
+ * subsetting.
+ */
+ @Test
+ public void testDisabledIcebergPredicateSubsetting() {
+ TQueryOptions queryOptions = new TQueryOptions();
+ queryOptions.setIceberg_predicate_pushdown_subsetting(false);
+ runPlannerTestFile("iceberg-predicates-disabled-subsetting", "functional_parquet",
+ queryOptions, ImmutableSet.of(PlannerTestOption.VALIDATE_CARDINALITY));
+ }
+
/**
* Check that Iceberg V2 table scans work as expected.
*/
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/iceberg-predicates-disabled-subsetting.test b/testdata/workloads/functional-planner/queries/PlannerTest/iceberg-predicates-disabled-subsetting.test
new file mode 100644
index 000000000..5421100c1
--- /dev/null
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/iceberg-predicates-disabled-subsetting.test
@@ -0,0 +1,32 @@
+# If one or more non-partition predicates are present, then all predicates are pushed down
+# to the scan node. ICEBERG_PREDICATE_PUSHDOWN_SUBSETTING query option is disabled for
+# these tests.
+select user from iceberg_partitioned where action = "download" and user = "Lisa";
+---- PLAN
+PLAN-ROOT SINK
+|
+00:SCAN HDFS [functional_parquet.iceberg_partitioned]
+ HDFS partitions=1/1 files=6 size=6.97KB
+ predicates: `user` = 'Lisa', action = 'download'
+ row-size=24B cardinality=1
+====
+# If no residual expression remain after Iceberg's planning, all partition-based
+# predicates can be skipped
+select * from iceberg_partitioned where action = "download";
+---- PLAN
+PLAN-ROOT SINK
+|
+00:SCAN HDFS [functional_parquet.iceberg_partitioned]
+ HDFS partitions=1/1 files=6 size=6.97KB
+ skipped Iceberg predicates: action = 'download'
+ row-size=44B cardinality=6
+====
+select * from iceberg_partitioned where action = "download" and event_time < "2022-01-01";
+---- PLAN
+PLAN-ROOT SINK
+|
+00:SCAN HDFS [functional_parquet.iceberg_partitioned]
+ HDFS partitions=1/1 files=6 size=6.97KB
+ skipped Iceberg predicates: action = 'download', event_time < TIMESTAMP '2022-01-01 00:00:00'
+ row-size=44B cardinality=1
+====
\ No newline at end of file
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/iceberg-predicates.test b/testdata/workloads/functional-planner/queries/PlannerTest/iceberg-predicates.test
index a682813a0..f3892e23f 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/iceberg-predicates.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/iceberg-predicates.test
@@ -34,22 +34,21 @@ PLAN-ROOT SINK
skipped Iceberg predicates: `year` = 2010
row-size=20B cardinality=730
====
-# Here both predicates are pushed to Iceberg and also to Impala's scan node. However,
-# here is a room for optimisation as we could skip pushing down 'year' to the scan node
-# as it won't filter further rows.
+# Here both predicates are pushed to Iceberg and only one is pushed to Impala's scan node,
+# 'year' predicate is filtered out, as it won't filter further rows
SELECT id, int_col, string_col from iceberg_partition_evolution where year = 2010 and id > 1000;
---- PLAN
PLAN-ROOT SINK
|
00:SCAN HDFS [functional_parquet.iceberg_partition_evolution]
HDFS partitions=1/1 files=730 size=1.25MB
- predicates: `year` = 2010, id > 1000
- row-size=24B cardinality=730
+ predicates: id > 1000
+ skipped Iceberg predicates: `year` = 2010
+ row-size=20B cardinality=730
====
# If we have predicates on partition columns with non-identity transform that could not
# be pushed to Iceberg then all the predicates are also pushed to Impala's scan node.
-# However, here is a room for optimisation as we could skip pushing down 'year' to the
-# scan node as it won't filter further rows.
+# 'year' predicate is filtered out, as it won't filter further rows
SELECT * FROM iceberg_partition_evolution
WHERE year = 2010 AND date_string_col='061610';
---- PLAN
@@ -57,7 +56,8 @@ PLAN-ROOT SINK
|
00:SCAN HDFS [functional_parquet.iceberg_partition_evolution]
HDFS partitions=1/1 files=2 size=3.49KB
- predicates: `year` = 2010, date_string_col = '061610'
+ predicates: date_string_col = '061610'
+ skipped Iceberg predicates: `year` = 2010
row-size=40B cardinality=2
====
# Checks when all the predicates are skipped in a count(*) query then the relevant
@@ -75,3 +75,95 @@ PLAN-ROOT SINK
skipped Iceberg predicates: action = 'click'
row-size=8B cardinality=6
====
+# List of predicates contains an untranslated expression: user = NULL.
+# (Iceberg predicate conversion can't handle ref = null expressions, it will result an
+# untranslated expression which must be pushed down to Impala's scanner.)
+select user from iceberg_partitioned where action = 'click' and user = null
+---- PLAN
+PLAN-ROOT SINK
+|
+00:SCAN HDFS [functional_parquet.iceberg_partitioned]
+ HDFS partitions=1/1 files=6 size=6.85KB
+ predicates: `user` = NULL
+ skipped Iceberg predicates: action = 'click'
+ row-size=12B cardinality=1
+---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
+01:EXCHANGE [UNPARTITIONED]
+|
+00:SCAN HDFS [functional_parquet.iceberg_partitioned]
+ HDFS partitions=1/1 files=6 size=6.85KB
+ predicates: `user` = NULL
+ skipped Iceberg predicates: action = 'click'
+ row-size=12B cardinality=1
+====
+# List of predicates contains an untranslated expression (action = NULL)
+select * from iceberg_partitioned where action = NULL and event_time < "2022-01-01";
+---- PLAN
+PLAN-ROOT SINK
+|
+00:SCAN HDFS [functional_parquet.iceberg_partitioned]
+ HDFS partitions=1/1 files=20 size=22.90KB
+ predicates: action = NULL
+ skipped Iceberg predicates: event_time < TIMESTAMP '2022-01-01 00:00:00'
+ row-size=44B cardinality=2
+====
+# List of predicates contains an untranslated expression (action LIKE "d%") and a redisual
+# expression after Iceberg's filtering
+select * from iceberg_partitioned where action like "d%" and event_time < "2022-01-01" and id < 10
+---- PLAN
+PLAN-ROOT SINK
+|
+00:SCAN HDFS [functional_parquet.iceberg_partitioned]
+ HDFS partitions=1/1 files=9 size=10.33KB
+ predicates: id < 10, action LIKE 'd%'
+ skipped Iceberg predicates: event_time < TIMESTAMP '2022-01-01 00:00:00'
+ row-size=44B cardinality=1
+====
+# Compound expression partially evaluated by Iceberg, and cannot be mapped back to Impala expression
+select * from iceberg_partitioned where action like "d%" and (event_time < "2020-01-01" or id > 10)
+---- PLAN
+PLAN-ROOT SINK
+|
+00:SCAN HDFS [functional_parquet.iceberg_partitioned]
+ HDFS partitions=1/1 files=10 size=11.42KB
+ predicates: (event_time < TIMESTAMP '2020-01-01 00:00:00' OR id > 10), action LIKE 'd%'
+ row-size=44B cardinality=1
+====
+# Predicate on a partition introduced by partition evolution pushed down to the scan node and
+# predicate on a partition that existed before partition evolution skipped.
+select * from iceberg_partition_evolution where month = 12 and year = 2010
+---- PLAN
+PLAN-ROOT SINK
+|
+00:SCAN HDFS [functional_parquet.iceberg_partition_evolution]
+ HDFS partitions=1/1 files=62 size=108.30KB
+ predicates: `month` = 12
+ skipped Iceberg predicates: `year` = 2010
+ row-size=40B cardinality=620
+====
+# Compound expression "(id > 5 or (id < 2))" returned as a residual expression, but separated
+# by the expression collector (IcebergExpressionCollector), mapping lookup fails, thus falling
+# back to the push down everything
+select * from iceberg_partitioned where action in ('click', 'view') and (id > 5 or (id < 2))
+---- PLAN
+PLAN-ROOT SINK
+|
+00:SCAN HDFS [functional_parquet.iceberg_partitioned]
+ HDFS partitions=1/1 files=12 size=13.65KB
+ predicates: action IN ('click', 'view'), (id > 5 OR (id < 2))
+ row-size=44B cardinality=1
+====
+# IS NOT NULL predicate skipped on partitioned column 'action', predicate on
+# non-partitioned 'id' column is pushed down.
+select * from iceberg_partitioned where action is not null and id < 10
+---- PLAN
+PLAN-ROOT SINK
+|
+00:SCAN HDFS [functional_parquet.iceberg_partitioned]
+ HDFS partitions=1/1 files=9 size=10.33KB
+ predicates: id < 10
+ skipped Iceberg predicates: action IS NOT NULL
+ row-size=44B cardinality=1
+====
\ No newline at end of file
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/iceberg-v2-tables-hash-join.test b/testdata/workloads/functional-planner/queries/PlannerTest/iceberg-v2-tables-hash-join.test
index 1ee7848d7..51b7b8c46 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/iceberg-v2-tables-hash-join.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/iceberg-v2-tables-hash-join.test
@@ -822,7 +822,8 @@ PLAN-ROOT SINK
|
00:SCAN HDFS [functional_parquet.iceberg_v2_partitioned_position_deletes]
HDFS partitions=1/1 files=1 size=1.17KB
- predicates: id > 0, action = 'download'
+ predicates: id > 0
+ skipped Iceberg predicates: action = 'download'
row-size=64B cardinality=1
---- DISTRIBUTEDPLAN
PLAN-ROOT SINK
@@ -840,7 +841,8 @@ PLAN-ROOT SINK
|
00:SCAN HDFS [functional_parquet.iceberg_v2_partitioned_position_deletes]
HDFS partitions=1/1 files=1 size=1.17KB
- predicates: id > 0, action = 'download'
+ predicates: id > 0
+ skipped Iceberg predicates: action = 'download'
row-size=64B cardinality=1
====
select * from iceberg_v2_partitioned_position_deletes
@@ -857,7 +859,8 @@ PLAN-ROOT SINK
|
00:SCAN HDFS [functional_parquet.iceberg_v2_partitioned_position_deletes]
HDFS partitions=1/1 files=1 size=1.17KB
- predicates: `user` = 'Lisa', action = 'download'
+ predicates: `user` = 'Lisa'
+ skipped Iceberg predicates: action = 'download'
row-size=64B cardinality=1
---- DISTRIBUTEDPLAN
PLAN-ROOT SINK
@@ -875,7 +878,8 @@ PLAN-ROOT SINK
|
00:SCAN HDFS [functional_parquet.iceberg_v2_partitioned_position_deletes]
HDFS partitions=1/1 files=1 size=1.17KB
- predicates: `user` = 'Lisa', action = 'download'
+ predicates: `user` = 'Lisa'
+ skipped Iceberg predicates: action = 'download'
row-size=64B cardinality=1
====
select event_time, action from iceberg_partitioned where (action = 'click' or action = 'view') and id > 0;
@@ -884,7 +888,8 @@ PLAN-ROOT SINK
|
00:SCAN HDFS [functional_parquet.iceberg_partitioned]
HDFS partitions=1/1 files=14 size=15.93KB
- predicates: id > 0, action IN ('click', 'view')
+ predicates: id > 0
+ skipped Iceberg predicates: action IN ('click', 'view')
row-size=32B cardinality=1
---- DISTRIBUTEDPLAN
PLAN-ROOT SINK
@@ -893,7 +898,8 @@ PLAN-ROOT SINK
|
00:SCAN HDFS [functional_parquet.iceberg_partitioned]
HDFS partitions=1/1 files=14 size=15.93KB
- predicates: id > 0, action IN ('click', 'view')
+ predicates: id > 0
+ skipped Iceberg predicates: action IN ('click', 'view')
row-size=32B cardinality=1
====
select event_time, action from iceberg_partitioned where action in ('click', 'view') and id > 0;
@@ -902,7 +908,8 @@ PLAN-ROOT SINK
|
00:SCAN HDFS [functional_parquet.iceberg_partitioned]
HDFS partitions=1/1 files=14 size=15.93KB
- predicates: id > 0, action IN ('click', 'view')
+ predicates: id > 0
+ skipped Iceberg predicates: action IN ('click', 'view')
row-size=32B cardinality=1
---- DISTRIBUTEDPLAN
PLAN-ROOT SINK
@@ -911,7 +918,8 @@ PLAN-ROOT SINK
|
00:SCAN HDFS [functional_parquet.iceberg_partitioned]
HDFS partitions=1/1 files=14 size=15.93KB
- predicates: id > 0, action IN ('click', 'view')
+ predicates: id > 0
+ skipped Iceberg predicates: action IN ('click', 'view')
row-size=32B cardinality=1
====
select event_time, action from iceberg_partitioned where (event_time='2020-01-01 11:00:00' or action = 'click') and id > 0;
@@ -920,7 +928,8 @@ PLAN-ROOT SINK
|
00:SCAN HDFS [functional_parquet.iceberg_partitioned]
HDFS partitions=1/1 files=6 size=6.85KB
- predicates: id > 0, (event_time = TIMESTAMP '2020-01-01 11:00:00' OR action = 'click')
+ predicates: id > 0
+ skipped Iceberg predicates: (event_time = TIMESTAMP '2020-01-01 11:00:00' OR action = 'click')
row-size=32B cardinality=1
---- DISTRIBUTEDPLAN
PLAN-ROOT SINK
@@ -929,7 +938,8 @@ PLAN-ROOT SINK
|
00:SCAN HDFS [functional_parquet.iceberg_partitioned]
HDFS partitions=1/1 files=6 size=6.85KB
- predicates: id > 0, (event_time = TIMESTAMP '2020-01-01 11:00:00' OR action = 'click')
+ predicates: id > 0
+ skipped Iceberg predicates: (event_time = TIMESTAMP '2020-01-01 11:00:00' OR action = 'click')
row-size=32B cardinality=1
====
select event_time, action from iceberg_partitioned where (event_time='2020-01-01 11:00:00' or action = 'click' or action = 'view') and id > 0;
@@ -938,7 +948,8 @@ PLAN-ROOT SINK
|
00:SCAN HDFS [functional_parquet.iceberg_partitioned]
HDFS partitions=1/1 files=14 size=15.93KB
- predicates: id > 0, (event_time = TIMESTAMP '2020-01-01 11:00:00' OR action = 'click' OR action = 'view')
+ predicates: id > 0
+ skipped Iceberg predicates: (event_time = TIMESTAMP '2020-01-01 11:00:00' OR action = 'click' OR action = 'view')
row-size=32B cardinality=1
---- DISTRIBUTEDPLAN
PLAN-ROOT SINK
@@ -947,7 +958,8 @@ PLAN-ROOT SINK
|
00:SCAN HDFS [functional_parquet.iceberg_partitioned]
HDFS partitions=1/1 files=14 size=15.93KB
- predicates: id > 0, (event_time = TIMESTAMP '2020-01-01 11:00:00' OR action = 'click' OR action = 'view')
+ predicates: id > 0
+ skipped Iceberg predicates: (event_time = TIMESTAMP '2020-01-01 11:00:00' OR action = 'click' OR action = 'view')
row-size=32B cardinality=1
====
select event_time, action from iceberg_partitioned where (event_time='2020-01-01 11:00:00' or action in ('click', 'view')) and id > 0;
@@ -956,7 +968,8 @@ PLAN-ROOT SINK
|
00:SCAN HDFS [functional_parquet.iceberg_partitioned]
HDFS partitions=1/1 files=14 size=15.93KB
- predicates: id > 0, (event_time = TIMESTAMP '2020-01-01 11:00:00' OR action IN ('click', 'view'))
+ predicates: id > 0
+ skipped Iceberg predicates: (event_time = TIMESTAMP '2020-01-01 11:00:00' OR action IN ('click', 'view'))
row-size=32B cardinality=1
---- DISTRIBUTEDPLAN
PLAN-ROOT SINK
@@ -965,7 +978,8 @@ PLAN-ROOT SINK
|
00:SCAN HDFS [functional_parquet.iceberg_partitioned]
HDFS partitions=1/1 files=14 size=15.93KB
- predicates: id > 0, (event_time = TIMESTAMP '2020-01-01 11:00:00' OR action IN ('click', 'view'))
+ predicates: id > 0
+ skipped Iceberg predicates: (event_time = TIMESTAMP '2020-01-01 11:00:00' OR action IN ('click', 'view'))
row-size=32B cardinality=1
====
select event_time, action from iceberg_partitioned where (event_time='2020-01-01 11:00:00' or action > 'a') and id > 0;
@@ -974,7 +988,8 @@ PLAN-ROOT SINK
|
00:SCAN HDFS [functional_parquet.iceberg_partitioned]
HDFS partitions=1/1 files=20 size=22.90KB
- predicates: id > 0, (event_time = TIMESTAMP '2020-01-01 11:00:00' OR action > 'a')
+ predicates: id > 0
+ skipped Iceberg predicates: (event_time = TIMESTAMP '2020-01-01 11:00:00' OR action > 'a')
row-size=32B cardinality=2
---- DISTRIBUTEDPLAN
PLAN-ROOT SINK
@@ -983,7 +998,8 @@ PLAN-ROOT SINK
|
00:SCAN HDFS [functional_parquet.iceberg_partitioned]
HDFS partitions=1/1 files=20 size=22.90KB
- predicates: id > 0, (event_time = TIMESTAMP '2020-01-01 11:00:00' OR action > 'a')
+ predicates: id > 0
+ skipped Iceberg predicates: (event_time = TIMESTAMP '2020-01-01 11:00:00' OR action > 'a')
row-size=32B cardinality=2
====
# All predicates are pushed down to Iceberg and won't filter any further rows. Skip pushing it to Scan node.
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/iceberg-v2-tables.test b/testdata/workloads/functional-planner/queries/PlannerTest/iceberg-v2-tables.test
index c0f31b79a..00360ecdb 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/iceberg-v2-tables.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/iceberg-v2-tables.test
@@ -822,7 +822,8 @@ PLAN-ROOT SINK
|
00:SCAN HDFS [functional_parquet.iceberg_v2_partitioned_position_deletes]
HDFS partitions=1/1 files=1 size=1.17KB
- predicates: id > 0, action = 'download'
+ predicates: id > 0
+ skipped Iceberg predicates: action = 'download'
row-size=64B cardinality=1
---- DISTRIBUTEDPLAN
PLAN-ROOT SINK
@@ -840,7 +841,8 @@ PLAN-ROOT SINK
|
00:SCAN HDFS [functional_parquet.iceberg_v2_partitioned_position_deletes]
HDFS partitions=1/1 files=1 size=1.17KB
- predicates: id > 0, action = 'download'
+ predicates: id > 0
+ skipped Iceberg predicates: action = 'download'
row-size=64B cardinality=1
====
select * from iceberg_v2_partitioned_position_deletes
@@ -857,7 +859,8 @@ PLAN-ROOT SINK
|
00:SCAN HDFS [functional_parquet.iceberg_v2_partitioned_position_deletes]
HDFS partitions=1/1 files=1 size=1.17KB
- predicates: `user` = 'Lisa', action = 'download'
+ predicates: `user` = 'Lisa'
+ skipped Iceberg predicates: action = 'download'
row-size=64B cardinality=1
---- DISTRIBUTEDPLAN
PLAN-ROOT SINK
@@ -875,7 +878,8 @@ PLAN-ROOT SINK
|
00:SCAN HDFS [functional_parquet.iceberg_v2_partitioned_position_deletes]
HDFS partitions=1/1 files=1 size=1.17KB
- predicates: `user` = 'Lisa', action = 'download'
+ predicates: `user` = 'Lisa'
+ skipped Iceberg predicates: action = 'download'
row-size=64B cardinality=1
====
select event_time, action from iceberg_partitioned where (action = 'click' or action = 'view') and id > 0;
@@ -884,7 +888,8 @@ PLAN-ROOT SINK
|
00:SCAN HDFS [functional_parquet.iceberg_partitioned]
HDFS partitions=1/1 files=14 size=15.93KB
- predicates: id > 0, action IN ('click', 'view')
+ predicates: id > 0
+ skipped Iceberg predicates: action IN ('click', 'view')
row-size=32B cardinality=1
---- DISTRIBUTEDPLAN
PLAN-ROOT SINK
@@ -893,7 +898,8 @@ PLAN-ROOT SINK
|
00:SCAN HDFS [functional_parquet.iceberg_partitioned]
HDFS partitions=1/1 files=14 size=15.93KB
- predicates: id > 0, action IN ('click', 'view')
+ predicates: id > 0
+ skipped Iceberg predicates: action IN ('click', 'view')
row-size=32B cardinality=1
====
select event_time, action from iceberg_partitioned where action in ('click', 'view') and id > 0;
@@ -902,7 +908,8 @@ PLAN-ROOT SINK
|
00:SCAN HDFS [functional_parquet.iceberg_partitioned]
HDFS partitions=1/1 files=14 size=15.93KB
- predicates: id > 0, action IN ('click', 'view')
+ predicates: id > 0
+ skipped Iceberg predicates: action IN ('click', 'view')
row-size=32B cardinality=1
---- DISTRIBUTEDPLAN
PLAN-ROOT SINK
@@ -911,7 +918,8 @@ PLAN-ROOT SINK
|
00:SCAN HDFS [functional_parquet.iceberg_partitioned]
HDFS partitions=1/1 files=14 size=15.93KB
- predicates: id > 0, action IN ('click', 'view')
+ predicates: id > 0
+ skipped Iceberg predicates: action IN ('click', 'view')
row-size=32B cardinality=1
====
select event_time, action from iceberg_partitioned where (event_time='2020-01-01 11:00:00' or action = 'click') and id > 0;
@@ -920,7 +928,8 @@ PLAN-ROOT SINK
|
00:SCAN HDFS [functional_parquet.iceberg_partitioned]
HDFS partitions=1/1 files=6 size=6.85KB
- predicates: id > 0, (event_time = TIMESTAMP '2020-01-01 11:00:00' OR action = 'click')
+ predicates: id > 0
+ skipped Iceberg predicates: (event_time = TIMESTAMP '2020-01-01 11:00:00' OR action = 'click')
row-size=32B cardinality=1
---- DISTRIBUTEDPLAN
PLAN-ROOT SINK
@@ -929,7 +938,8 @@ PLAN-ROOT SINK
|
00:SCAN HDFS [functional_parquet.iceberg_partitioned]
HDFS partitions=1/1 files=6 size=6.85KB
- predicates: id > 0, (event_time = TIMESTAMP '2020-01-01 11:00:00' OR action = 'click')
+ predicates: id > 0
+ skipped Iceberg predicates: (event_time = TIMESTAMP '2020-01-01 11:00:00' OR action = 'click')
row-size=32B cardinality=1
====
select event_time, action from iceberg_partitioned where (event_time='2020-01-01 11:00:00' or action = 'click' or action = 'view') and id > 0;
@@ -938,7 +948,8 @@ PLAN-ROOT SINK
|
00:SCAN HDFS [functional_parquet.iceberg_partitioned]
HDFS partitions=1/1 files=14 size=15.93KB
- predicates: id > 0, (event_time = TIMESTAMP '2020-01-01 11:00:00' OR action = 'click' OR action = 'view')
+ predicates: id > 0
+ skipped Iceberg predicates: (event_time = TIMESTAMP '2020-01-01 11:00:00' OR action = 'click' OR action = 'view')
row-size=32B cardinality=1
---- DISTRIBUTEDPLAN
PLAN-ROOT SINK
@@ -947,7 +958,8 @@ PLAN-ROOT SINK
|
00:SCAN HDFS [functional_parquet.iceberg_partitioned]
HDFS partitions=1/1 files=14 size=15.93KB
- predicates: id > 0, (event_time = TIMESTAMP '2020-01-01 11:00:00' OR action = 'click' OR action = 'view')
+ predicates: id > 0
+ skipped Iceberg predicates: (event_time = TIMESTAMP '2020-01-01 11:00:00' OR action = 'click' OR action = 'view')
row-size=32B cardinality=1
====
select event_time, action from iceberg_partitioned where (event_time='2020-01-01 11:00:00' or action in ('click', 'view')) and id > 0;
@@ -956,7 +968,8 @@ PLAN-ROOT SINK
|
00:SCAN HDFS [functional_parquet.iceberg_partitioned]
HDFS partitions=1/1 files=14 size=15.93KB
- predicates: id > 0, (event_time = TIMESTAMP '2020-01-01 11:00:00' OR action IN ('click', 'view'))
+ predicates: id > 0
+ skipped Iceberg predicates: (event_time = TIMESTAMP '2020-01-01 11:00:00' OR action IN ('click', 'view'))
row-size=32B cardinality=1
---- DISTRIBUTEDPLAN
PLAN-ROOT SINK
@@ -965,7 +978,8 @@ PLAN-ROOT SINK
|
00:SCAN HDFS [functional_parquet.iceberg_partitioned]
HDFS partitions=1/1 files=14 size=15.93KB
- predicates: id > 0, (event_time = TIMESTAMP '2020-01-01 11:00:00' OR action IN ('click', 'view'))
+ predicates: id > 0
+ skipped Iceberg predicates: (event_time = TIMESTAMP '2020-01-01 11:00:00' OR action IN ('click', 'view'))
row-size=32B cardinality=1
====
select event_time, action from iceberg_partitioned where (event_time='2020-01-01 11:00:00' or action > 'a') and id > 0;
@@ -974,7 +988,8 @@ PLAN-ROOT SINK
|
00:SCAN HDFS [functional_parquet.iceberg_partitioned]
HDFS partitions=1/1 files=20 size=22.90KB
- predicates: id > 0, (event_time = TIMESTAMP '2020-01-01 11:00:00' OR action > 'a')
+ predicates: id > 0
+ skipped Iceberg predicates: (event_time = TIMESTAMP '2020-01-01 11:00:00' OR action > 'a')
row-size=32B cardinality=2
---- DISTRIBUTEDPLAN
PLAN-ROOT SINK
@@ -983,7 +998,8 @@ PLAN-ROOT SINK
|
00:SCAN HDFS [functional_parquet.iceberg_partitioned]
HDFS partitions=1/1 files=20 size=22.90KB
- predicates: id > 0, (event_time = TIMESTAMP '2020-01-01 11:00:00' OR action > 'a')
+ predicates: id > 0
+ skipped Iceberg predicates: (event_time = TIMESTAMP '2020-01-01 11:00:00' OR action > 'a')
row-size=32B cardinality=2
====
# All predicates are pushed down to Iceberg and won't filter any further rows. Skip pushing it to Scan node.
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/tablesample.test b/testdata/workloads/functional-planner/queries/PlannerTest/tablesample.test
index 3193f0254..795e7ff79 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/tablesample.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/tablesample.test
@@ -323,13 +323,14 @@ PLAN-ROOT SINK
|
00:SCAN HDFS [functional_parquet.iceberg_partitioned]
HDFS partitions=1/1 files=4 size=4.57KB
- predicates: id > CAST(0 AS INT), action = 'click'
+ predicates: id > CAST(0 AS INT)
+ skipped Iceberg predicates: action = 'click'
stored statistics:
table: rows=20 size=22.90KB
columns: unavailable
extrapolated-rows=disabled max-scan-range-rows=5
- parquet statistics predicates: id > CAST(0 AS INT), action = 'click'
- parquet dictionary predicates: id > CAST(0 AS INT), action = 'click'
+ parquet statistics predicates: id > CAST(0 AS INT)
+ parquet dictionary predicates: id > CAST(0 AS INT)
mem-estimate=64.00MB mem-reservation=32.00KB thread-reservation=1
tuple-ids=0 row-size=44B cardinality=1
in pipelines: 00(GETNEXT)