jprdonnelly/impala
1
0
Fork 0
mirror of https://github.com/apache/impala.git synced 2026-01-31 00:00:20 -05:00
Code Issues Projects Releases Wiki Activity

IMPALA-10898: Add runtime IN-list filters for ORC tables

Browse Source 
ORC files have optional bloom filter indexes for each column. Since
ORC-1.7.0, the C++ reader supports pushing down predicates to skip
unreleated RowGroups. The pushed down predicates will be evaludated on
file indexes (i.e. statistics and bloom filter indexes). Note that only
EQUALS and IN-list predicates can leverage bloom filter indexes.

Currently Impala has two kinds of runtime filters: bloom filter and
min-max filter. Unfortunately they can't be converted into EQUALS or
IN-list predicates. So they can't leverage the file level bloom filter
indexes.

This patch adds runtime IN-list filters for this purpose. Currently they
are generated for the build side of a broadcast join. They will only be
applied on ORC tables and be pushed down to the ORC reader(i.e. ORC
lib). To avoid exploding the IN-list, if # of distinct values of the
build side exceeds a threshold (default to 1024), we set the filter to
ALWAYS_TRUE and clear its entry. The threshold can be configured by a
new query option, RUNTIME_IN_LIST_FILTER_ENTRY_LIMIT.

Evaluating runtime IN-list filters is much slower than evaluating
runtime bloom filters due to the current simple implementation (i.e.
std::unorder_set) and the lack of codegen. So we disable it at row
level.

For visibility, this patch addes two counters in the HdfsScanNode:
 - NumPushedDownPredicates
 - NumPushedDownRuntimeFilters
They reflect the predicates and runtime filters that are pushed down to
the ORC reader.

Currently, runtime IN-list filters are disabled by default. This patch
extends the query option, ENABLED_RUNTIME_FILTER_TYPES, to support a
comma separated list of filter types. It defaults to be "BLOOM,MIN_MAX".
Add "IN_LIST" in it to enable runtime IN-list filters.

Ran perf tests on a 3 instances cluster on my desktop using TPC-DS with
scale factor 20. It shows significant improvements in some queries:

+-----------+-------------+--------------------+--------+-------------+------------+------------+----------------+-------+----------------+---------+--------+
| Workload  | Query       | File Format        | Avg(s) | Base Avg(s) | Delta(Avg) | StdDev(%)  | Base StdDev(%) | Iters | Median Diff(%) | MW Zval | Tval   |
+-----------+-------------+--------------------+--------+-------------+------------+------------+----------------+-------+----------------+---------+--------+
| TPCDS(20) | TPCDS-Q67A  | orc / snap / block | 35.07  | 44.01       | I -20.32%  |   0.38%    |   1.38%        | 10    | I -25.69%      | -3.58   | -45.33 |
| TPCDS(20) | TPCDS-Q37   | orc / snap / block | 1.08   | 1.45        | I -25.23%  |   7.14%    |   3.09%        | 10    | I -34.09%      | -3.58   | -12.94 |
| TPCDS(20) | TPCDS-Q70A  | orc / snap / block | 6.30   | 8.60        | I -26.81%  |   5.24%    |   4.21%        | 10    | I -36.67%      | -3.58   | -14.88 |
| TPCDS(20) | TPCDS-Q16   | orc / snap / block | 1.33   | 1.85        | I -28.28%  |   4.98%    |   5.92%        | 10    | I -39.38%      | -3.58   | -12.93 |
| TPCDS(20) | TPCDS-Q18A  | orc / snap / block | 5.70   | 8.06        | I -29.25%  |   3.00%    |   4.12%        | 10    | I -40.30%      | -3.58   | -19.95 |
| TPCDS(20) | TPCDS-Q22A  | orc / snap / block | 2.01   | 2.97        | I -32.21%  |   6.12%    |   5.94%        | 10    | I -47.68%      | -3.58   | -14.05 |
| TPCDS(20) | TPCDS-Q77A  | orc / snap / block | 8.49   | 12.44       | I -31.75%  |   6.44%    |   3.96%        | 10    | I -49.71%      | -3.58   | -16.97 |
| TPCDS(20) | TPCDS-Q75   | orc / snap / block | 7.76   | 12.27       | I -36.76%  |   5.01%    |   3.87%        | 10    | I -59.56%      | -3.58   | -23.26 |
| TPCDS(20) | TPCDS-Q21   | orc / snap / block | 0.71   | 1.27        | I -44.26%  |   4.56%    |   4.24%        | 10    | I -77.31%      | -3.58   | -28.31 |
| TPCDS(20) | TPCDS-Q80A  | orc / snap / block | 9.24   | 20.42       | I -54.77%  |   4.03%    |   3.82%        | 10    | I -123.12%     | -3.58   | -40.90 |
| TPCDS(20) | TPCDS-Q39-1 | orc / snap / block | 1.07   | 2.26        | I -52.74%  | * 23.83% * |   2.60%        | 10    | I -149.68%     | -3.58   | -14.43 |
| TPCDS(20) | TPCDS-Q39-2 | orc / snap / block | 1.00   | 2.33        | I -56.95%  | * 19.53% * |   2.07%        | 10    | I -151.89%     | -3.58   | -20.81 |
+-----------+-------------+--------------------+--------+-------------+------------+------------+----------------+-------+----------------+---------+--------+
"Base Avg" is the avg of the original time. "Avg" is the current time.

However, we also see some regressions due to the suboptimal
implementation. The follow-up JIRAs will focus on improvements:
 - IMPALA-11140: Codegen InListFilter::Insert() and InListFilter::Find()
 - IMPALA-11141: Use exact data types in IN-list filters instead of
   casting data to a set of int64_t or a set of string.
 - IMPALA-11142: Consider IN-list filters in partitioned joins.

Tests:
 - Test IN-list filter on string, date and all integer types
 - Test IN-list filter with NULL
 - Test IN-list filter on complex exprs targets

Change-Id: I25080628233799aa0b6be18d5a832f1385414501
Reviewed-on: http://gerrit.cloudera.org:8080/18141
Reviewed-by: Qifan Chen <qchen@cloudera.com>
Tested-by: Impala Public Jenkins <impala-public-jenkins@cloudera.com>
This commit is contained in:
stiga-huang
2021-12-01 15:23:03 +08:00
committed by Impala Public Jenkins
parent b2e4b29f06
commit 374783c55e
44 changed files with 1430 additions and 215 deletions
Download Patch File Download Diff File Expand all files Collapse all files

173
testdata/workloads/functional-query/queries/QueryTest/in_list_filters.test vendored Normal file
View File

@@ -0,0 +1,173 @@
====
---- QUERY
# Test IN-list filter on partition columns.
# There are 24 partitions and 24 files in alltypes. 22 of them will be filtered out.
# Expect 22 / 3 = 7 files be rejected per scan fragment.
select STRAIGHT_JOIN count(*) from alltypes p join [BROADCAST] alltypestiny b
on p.month = b.int_col and b.month = 1 and b.string_col = "1"
---- RESULTS
620
---- RUNTIME_PROFILE
row_regex: .*Filter 0 arrival with 1 items.*
row_regex: .*Files rejected: 7 \(7\).*
====
---- QUERY
# Test two hop IN-list filters on partition columns.
# "c.month = 13" won't match any rows in alltypestiny. Expect all files be rejected.
select STRAIGHT_JOIN count(*) from alltypes a
join [BROADCAST] alltypes b
join [BROADCAST] alltypestiny c
where c.month = 13 and b.year = c.year and a.month = b.month
---- RESULTS
0
---- RUNTIME_PROFILE
row_regex: .*Filter 0 arrival with 0 items.*
row_regex: .*Filter 1 arrival with 0 items.*
row_regex: .*Files rejected: 8 \(8\).*
====
---- QUERY
# Test IN-list filter on string column.
select STRAIGHT_JOIN count(*) from alltypes a
join [BROADCAST] alltypestiny_negative b
where a.string_col = b.string_col
---- RESULTS
0
---- RUNTIME_PROFILE
row_regex: .*Filter 0 arrival with 1 items.*
row_regex: .*NumPushedDownRuntimeFilters: 1 \(1\).*
row_regex: .*RowsRead: 0 \(0\).*
====
---- QUERY
# Test IN-list filter on empty strings.
select STRAIGHT_JOIN count(*) from alltypes a
join [BROADCAST] nulltable n
where a.string_col = n.b;
---- RESULTS
0
---- RUNTIME_PROFILE
row_regex: .*Filter 0 arrival with 1 items.*
row_regex: .*NumPushedDownRuntimeFilters: 1 \(1\).*
row_regex: .*RowsRead: 0 \(0\).*
====
---- QUERY
# Test IN-list filter on tinyint column.
select STRAIGHT_JOIN count(*) from alltypes a
join [BROADCAST] alltypestiny_negative b
where a.tinyint_col = b.tinyint_col
---- RESULTS
0
---- RUNTIME_PROFILE
row_regex: .*Filter 0 arrival with 1 items.*
row_regex: .*NumPushedDownRuntimeFilters: 1 \(1\).*
row_regex: .*RowsRead: 0 \(0\).*
====
---- QUERY
# Test IN-list filter on smallint column.
select STRAIGHT_JOIN count(*) from alltypes a
join [BROADCAST] alltypestiny_negative b
where a.smallint_col = b.smallint_col
---- RESULTS
0
---- RUNTIME_PROFILE
row_regex: .*Filter 0 arrival with 1 items.*
row_regex: .*NumPushedDownRuntimeFilters: 1 \(1\).*
row_regex: .*RowsRead: 0 \(0\).*
====
---- QUERY
# Test IN-list filter on int column.
select STRAIGHT_JOIN count(*) from alltypes a
join [BROADCAST] alltypestiny_negative b
where a.int_col = b.int_col
---- RESULTS
0
---- RUNTIME_PROFILE
row_regex: .*Filter 0 arrival with 1 items.*
row_regex: .*NumPushedDownRuntimeFilters: 1 \(1\).*
row_regex: .*RowsRead: 0 \(0\).*
====
---- QUERY
# Test IN-list filter on bigint column.
select STRAIGHT_JOIN count(*) from alltypes a
join [BROADCAST] alltypestiny_negative b
where a.bigint_col = b.bigint_col
---- RESULTS
0
---- RUNTIME_PROFILE
row_regex: .*Filter 0 arrival with 1 items.*
row_regex: .*NumPushedDownRuntimeFilters: 1 \(1\).*
row_regex: .*RowsRead: 0 \(0\).*
====
---- QUERY
# Test IN-list filter on bigint column.
select STRAIGHT_JOIN count(*) from alltypes a
join [BROADCAST] alltypestiny b
where a.bigint_col = b.bigint_col + 100
---- RESULTS
0
---- RUNTIME_PROFILE
row_regex: .*Filter 0 arrival with 2 items.*
row_regex: .*NumPushedDownRuntimeFilters: 1 \(1\).*
row_regex: .*RowsRead: 0 \(0\).*
====
---- QUERY
# Test IN-list filter on DATE partition column.
# 2 of the 4 partitions are filtered out.
select STRAIGHT_JOIN count(*) from date_tbl a
join [BROADCAST] date_tbl b
on a.date_part = b.date_col
---- RESULTS
11
---- RUNTIME_PROFILE
row_regex: .*Filter 0 arrival with 17 items.*
aggregation(SUM, Files rejected): 2
====
---- QUERY
# Test IN-list filter on DATE non-partition column.
select STRAIGHT_JOIN count(*) from date_tbl a
join [BROADCAST] date_tbl b
on a.date_col = b.date_part
---- RESULTS
11
---- RUNTIME_PROFILE
row_regex: .*Filter 0 arrival with 4 items.*
row_regex: .*NumPushedDownRuntimeFilters: 1 \(1\).*
====
---- QUERY
# Test IN-list filter with NULL.
# 'id' is a string column without NULLs. 'null_str' is a string column with all NULLs.
# The pushed down IN-list filter should be able to filter out all rows.
select STRAIGHT_JOIN count(*) from nullrows a
join [BROADCAST] nullrows b
where a.id <=> b.null_str;
---- RESULTS
0
---- RUNTIME_PROFILE
row_regex: .*Filter 0 arrival with 1 items.*
row_regex: .*NumPushedDownRuntimeFilters: 1 \(1\).*
row_regex: .*RowsRead: 0 \(0\).*
====
---- QUERY
# Test IN-list filter on complex target expr, i.e. not a simple slot ref.
# The filter can't be pushed down to the ORC lib since the ORC lib can't evaluate the
# expr. Expect 7300 / 3 = 2433 rows read per scan fragment on 'alltypes'.
select STRAIGHT_JOIN count(*) from functional_orc_def.alltypes a
join [BROADCAST] functional_orc_def.alltypestiny b
on a.id + 1 = b.id
---- RESULTS
7
---- RUNTIME_PROFILE
row_regex: .*RowsRead: 2.43K \(2433\).*
====
---- QUERY
# Test IN-list filter on wide string that exceeds the total string size.
# The filter is turned off (always_true=true). Expect it arrives with 0 items.
# Expect 7300 / 3 = 2433 rows read per scan fragment on 'alltypes'.
set max_row_size=16m;
select STRAIGHT_JOIN count(*) from alltypes a
join [BROADCAST] widerow b
on a.string_col = b.string_col
---- RUNTIME_PROFILE
row_regex: .*Filter 0 arrival with 0 items.*
row_regex: .*RowsRead: 2.43K \(2433\).*
====