Joining two very large Kudu tables in Impala

[Franco VENTURI <fv...@comcast.net>]

At work we have two very large tables stored in Kudu (since we need to be able to perform row updates efficiently).
One of them (called table33 in the example SQL below) contains two columns called 'join_column1' and 'join_column2' that are used to join it with the other table (called table14 below).


A typical query would look like this one:


SELECT STRAIGHT_JOIN
       d.pk_date1_column,
       d.columnA,
       d.columnB,
       d.columnC
  FROM schema.table14 d
  JOIN (
SELECT join_column1, join_column2 FROM schema.table33 dt
WHERE dt.pk_column = 1234567890
   AND dt.pk_date2_column >= date_trunc('second',to_timestamp('06-10-2019 18:13:48', 'MM-dd-yyyy HH:mm:ss'))
   AND dt.pk_date2_column <= seconds_add(date_trunc('second',to_timestamp('06-10-2019 18:13:48', 'MM-dd-yyyy HH:mm:ss')),1)
   AND dt.column3 = 'SOME_VALUE_FOR_COLUMN_3'
   AND dt.column4 = 'SOME_VALUE_FOR_COLUMN_4') AS dtdt
WHERE dtdt.join_column1 = d.join_column1
   AND dtdt.join_column2 = d.join_column2
   AND d.pk_date1_column >= to_date(date_trunc('month',months_sub(to_timestamp('06-10-2019 18:13:48', 'MM-dd-yyyy HH:mm:ss'), 3)))
   AND d.pk_date1_column <= to_date(date_trunc('month',months_add(to_timestamp('06-10-2019 18:13:48', 'MM-dd-yyyy HH:mm:ss'), 6)))
ORDER BY d.pk_date1_column, d.date2_column;


The idea here is to use a very narrow subselect (using the columns 'pk_column', 'column3', and 'column4' from table33) to just select very few rows (a dozen or so), and then join with the other table (table14) to return the values of 'columnA', 'columnB', and 'columnC' for these few rows - if we 'knew' the values of the join columns for these two rows in advance, the whole join could be easily be replaced by an IN-list.


Some other useful information:
  - the cardinality of table14 is about 14 billion rows
  - the cardinality of table33 is about 33 billion rows
  - the purpose of the 'STRAIGHT_JOIN' hint is to avoid to have the Impala optimizer reorder the join (and hence get rid of any 'COMPUTE STATS' discussion)
  - the SQL query above is roughly modeled after the example here: https://impala.apache.org/docs/build/html/topics/impala_perf_joins.html (section: 'Overriding Join Reordering with STRAIGHT_JOIN')
  - the version of Impala we are using is 2.12 and the version of Kudu we are using is 1.7.0 (this is what comes with our vendor distribution)


Running EXPLAIN on the example query above shows the following plan:


PLAN-ROOT SINK
|
05:MERGING-EXCHANGE [UNPARTITIONED]
|  order by: pk_date1_column ASC, date2_column ASC
|
03:SORT
|  order by: pk_date1_column ASC, date2_column ASC
|
02:HASH JOIN [INNER JOIN, BROADCAST]
|  hash predicates: d.join_column2 = join_column2, d.join_column1 = join_column1
|  runtime filters: RF002 <- join_column2, RF003 <- join_column1
|
|--04:EXCHANGE [BROADCAST]
|  |
|  01:SCAN KUDU [schema.table33 dt]
|     kudu predicates: dt.pk_column = 1234567890, dt.column3 = 'SOME_VALUE_FOR_COLUMN_3', dt.pk_date2_column <= TIMESTAMP '2019-06-10 18:13:49', dt.pk_date2_column >= TIMESTAMP '2019-06-10 18:13:48', dt.column4 = 'SOME_VALUE_FOR_COLUMN_4'
|
00:SCAN KUDU [schema.table14 d]
   kudu predicates: d.pk_date1_column <= TIMESTAMP '2019-12-01 00:00:00', d.pk_date1_column >= TIMESTAMP '2019-03-01 00:00:00'
   runtime filters: RF002 -> d.join_column2, RF003 -> d.join_column1


As you can see above, step '01:SCAN KUDU' uses the kudu predicates to filter down the results from table33 (and when the query runs, the query profile shows it takes only a few ms) and returns a dozen or so rows, however step '00:SCAN KUDU' doesn't use any of the results from the other part of the join and goes through billions of rows in table14 (taking about 10 minutes or more on our cluster) to finally return the columns requested in the main select.


On the other hand, if we manually run the very narrow subselect, cut&paste the join_column values for the few returned rows and replace the join with an IN-list, Impala takes advantage of this information and returns the requested columns in sub-second time.


This (joining two very large table to extract only a few rows) seems a very common use case for us and we were surprised to see this significant performance impact when using Impala on Kudu tables.


Going through the Impala JIRA we found a couple of interesting pieces of information:
  - a comment by Todd Lipcon (https://issues.apache.org/jira/browse/IMPALA-4252?focusedCommentId=15921058&page=com.atlassian.jira.plugin.system.issuetabpanels%3Acomment-tabpanel#comment-15921058) back in October 2016, suggesting the use of IN-lists, but it doesn't seem to have had any reply
  - a JIRA task about pushing Bloom filters to Kudu scanners (https://issues.apache.org/jira/browse/IMPALA-3741), created in June 2016, but it looks like it is still unresolved.


I was wondering if any of you had to face a similar issue joining two very large Kudu tables, and if you found an efficient way with Impala to run this kind of query (short of having to re-design those two tables, which would be a non-trivial task for us, given the table sizes, and given the number of other code changes it would require in all the other applications that populate and query those tables).


We could also tell all our end users that they can no longer have this kind of two table joins (which they have been used to for years in our current RDBMS), and that instead they have to run two selects and then add some code for the IN-list logic, but we were hoping first to be able to find an elegant SQL-only solution to this problem.


Thanks in advance,
Franco Venturi