Uploaded image for project: 'IMPALA'
  1. IMPALA
  2. IMPALA-3573

Planner doesn't take into account runtime filter selectivity

    Details

    • Type: Improvement
    • Status: Open
    • Priority: Major
    • Resolution: Unresolved
    • Affects Version/s: Impala 2.5.0
    • Fix Version/s: None
    • Component/s: Frontend
    • Labels:

      Description

      Applying selective runtime filters can drastically change the cardinality of scan nodes, the planner doesn't cost the runtime filters as filters as a result it misses out on a more selective plan.

      In TPC-DS query29 there are three fact to dimension joins

      • (ss x d1) -> 389.28M rows
      • (sr x d2) -> 234.43M rows
      • (cs x d3) -> 12.85B rows

      The planner doesn't re-evaluate the cardinality estimation of ss, sr and cs after the runtime filter are applied and puts ss as the left most node in the plan where it should have been cs.

      Ideally this should be a bushy plan....

      Query

      select i_item_id
          ,i_item_desc
          ,s_store_id
          ,s_store_name
          ,sum(ss_quantity)        as store_sales_quantity
          ,sum(sr_return_quantity) as store_returns_quantity
          ,sum(cs_quantity)        as catalog_sales_quantity
       from
          store_sales
         ,store_returns
         ,catalog_sales
         ,date_dim             d1
         ,date_dim             d2
         ,date_dim             d3
         ,store
         ,item
       where
           d1.d_moy               = 4 
       and d1.d_year              = 1999
       and d1.d_date_sk           = ss_sold_date_sk
       and i_item_sk              = ss_item_sk
       and s_store_sk             = ss_store_sk
       and ss_customer_sk         = sr_customer_sk
       and ss_item_sk             = sr_item_sk
       and ss_ticket_number       = sr_ticket_number
       and sr_returned_date_sk    = d2.d_date_sk
       and d2.d_moy               between 4 and  4 + 3 
       and d2.d_year              = 1999
       and sr_customer_sk         = cs_bill_customer_sk
       and sr_item_sk             = cs_item_sk
       and cs_sold_date_sk        = d3.d_date_sk     
       and d3.d_year              in (1999,1999+1,1999+2)
       group by
          i_item_id
         ,i_item_desc
         ,s_store_id
         ,s_store_name
       order by
          i_item_id 
         ,i_item_desc
         ,s_store_id
         ,s_store_name
      limit 100
      

      Plan

      28:MERGING-EXCHANGE [UNPARTITIONED]
      |  order by: i_item_id ASC, i_item_desc ASC, s_store_id ASC, s_store_name ASC
      |  limit: 100
      |  hosts=20 per-host-mem=unavailable
      |  tuple-ids=9 row-size=224B cardinality=100
      |
      16:TOP-N [LIMIT=100]
      |  order by: i_item_id ASC, i_item_desc ASC, s_store_id ASC, s_store_name ASC
      |  hosts=20 per-host-mem=21.89KB
      |  tuple-ids=9 row-size=224B cardinality=100
      |
      27:AGGREGATE [FINALIZE]
      |  output: sum:merge(ss_quantity), sum:merge(sr_return_quantity), sum:merge(cs_quantity)
      |  group by: i_item_id, i_item_desc, s_store_id, s_store_name
      |  hosts=20 per-host-mem=212.22MB
      |  tuple-ids=8 row-size=224B cardinality=335852270
      |
      26:EXCHANGE [HASH(i_item_id,i_item_desc,s_store_id,s_store_name)]
      |  hosts=20 per-host-mem=0B
      |  tuple-ids=8 row-size=224B cardinality=335852270
      |
      15:AGGREGATE [STREAMING]
      |  output: sum(ss_quantity), sum(sr_return_quantity), sum(cs_quantity)
      |  group by: i_item_id, i_item_desc, s_store_id, s_store_name
      |  hosts=20 per-host-mem=77.13GB
      |  tuple-ids=8 row-size=224B cardinality=335852270
      |
      14:HASH JOIN [INNER JOIN, BROADCAST]
      |  hash predicates: ss_item_sk = i_item_sk
      |  runtime filters: RF000 <- i_item_sk
      |  hosts=20 per-host-mem=5.24MB
      |  tuple-ids=0,3,1,4,2,5,6,7 row-size=368B cardinality=335852270
      |
      |--25:EXCHANGE [BROADCAST]
      |  |  hosts=1 per-host-mem=0B
      |  |  tuple-ids=7 row-size=156B cardinality=32000
      |  |
      |  07:SCAN HDFS [tpcds_15000_decimal_parquet.item, RANDOM]
      |     partitions=1/1 files=1 size=3.14MB
      |     table stats: 32000 rows total
      |     column stats: all
      |     hosts=1 per-host-mem=48.00MB
      |     tuple-ids=7 row-size=156B cardinality=32000
      |
      13:HASH JOIN [INNER JOIN, BROADCAST]
      |  hash predicates: ss_store_sk = s_store_sk
      |  runtime filters: RF001 <- s_store_sk
      |  hosts=20 per-host-mem=4.00KB
      |  tuple-ids=0,3,1,4,2,5,6 row-size=212B cardinality=335852270
      |
      |--24:EXCHANGE [BROADCAST]
      |  |  hosts=1 per-host-mem=0B
      |  |  tuple-ids=6 row-size=60B cardinality=62
      |  |
      |  06:SCAN HDFS [tpcds_15000_decimal_parquet.store, RANDOM]
      |     partitions=1/1 files=1 size=11.92KB
      |     table stats: 62 rows total
      |     column stats: all
      |     hosts=1 per-host-mem=48.00MB
      |     tuple-ids=6 row-size=60B cardinality=62
      |
      12:HASH JOIN [INNER JOIN, BROADCAST]
      |  hash predicates: cs_sold_date_sk = d3.d_date_sk
      |  runtime filters: RF002 <- d3.d_date_sk
      |  hosts=20 per-host-mem=25.23KB
      |  tuple-ids=0,3,1,4,2,5 row-size=152B cardinality=335852270
      |
      |--23:EXCHANGE [BROADCAST]
      |  |  hosts=1 per-host-mem=0B
      |  |  tuple-ids=5 row-size=12B cardinality=1957
      |  |
      |  05:SCAN HDFS [tpcds_15000_decimal_parquet.date_dim d3, RANDOM]
      |     partitions=1/1 files=1 size=2.17MB
      |     predicates: d3.d_year IN (1999, 1999 + 1, 1999 + 2)
      |     table stats: 73049 rows total
      |     column stats: all
      |     hosts=1 per-host-mem=32.00MB
      |     tuple-ids=5 row-size=12B cardinality=1957
      |
      11:HASH JOIN [INNER JOIN, PARTITIONED]
      |  hash predicates: sr_customer_sk = cs_bill_customer_sk, sr_item_sk = cs_item_sk
      |  runtime filters: RF003 <- cs_bill_customer_sk, RF004 <- cs_item_sk
      |  hosts=20 per-host-mem=30.98GB
      |  tuple-ids=0,3,1,4,2 row-size=140B cardinality=352496606
      |
      |--22:EXCHANGE [HASH(cs_bill_customer_sk,cs_item_sk)]
      |  |  hosts=20 per-host-mem=0B
      |  |  tuple-ids=2 row-size=28B cardinality=21602442972
      |  |
      |  02:SCAN HDFS [tpcds_15000_decimal_parquet.catalog_sales, RANDOM]
      |     partitions=1837/1837 files=6122 size=1.33TB
      |     runtime filters: RF000 -> tpcds_15000_decimal_parquet.catalog_sales.cs_item_sk, RF002 -> cs_sold_date_sk
      |     table stats: 21602442972 rows total
      |     column stats: all
      |     hosts=20 per-host-mem=264.00MB
      |     tuple-ids=2 row-size=28B cardinality=21602442972
      |
      21:EXCHANGE [HASH(sr_customer_sk,sr_item_sk)]
      |  hosts=20 per-host-mem=0B
      |  tuple-ids=0,3,1,4 row-size=112B cardinality=352496606
      |
      10:HASH JOIN [INNER JOIN, BROADCAST]
      |  hash predicates: sr_returned_date_sk = d2.d_date_sk
      |  runtime filters: RF005 <- d2.d_date_sk
      |  hosts=20 per-host-mem=3.54KB
      |  tuple-ids=0,3,1,4 row-size=112B cardinality=352496606
      |
      |--20:EXCHANGE [BROADCAST]
      |  |  hosts=1 per-host-mem=0B
      |  |  tuple-ids=4 row-size=16B cardinality=206
      |  |
      |  04:SCAN HDFS [tpcds_15000_decimal_parquet.date_dim d2, RANDOM]
      |     partitions=1/1 files=1 size=2.17MB
      |     predicates: d2.d_year = 1999, d2.d_moy >= 4, d2.d_moy <= 4 + 3
      |     table stats: 73049 rows total
      |     column stats: all
      |     hosts=1 per-host-mem=48.00MB
      |     tuple-ids=4 row-size=16B cardinality=206
      |
      09:HASH JOIN [INNER JOIN, PARTITIONED]
      |  hash predicates: ss_customer_sk = sr_customer_sk, ss_item_sk = sr_item_sk, ss_ticket_number = sr_ticket_number
      |  runtime filters: RF006 <- sr_customer_sk, RF007 <- sr_item_sk, RF008 <- sr_ticket_number
      |  hosts=20 per-host-mem=7.96GB
      |  tuple-ids=0,3,1 row-size=96B cardinality=3834191129
      |
      |--19:EXCHANGE [HASH(sr_customer_sk,sr_item_sk,sr_ticket_number)]
      |  |  hosts=20 per-host-mem=0B
      |  |  tuple-ids=1 row-size=36B cardinality=4317083368
      |  |
      |  01:SCAN HDFS [tpcds_15000_decimal_parquet.store_returns, RANDOM]
      |     partitions=2004/2004 files=2022 size=201.98GB
      |     runtime filters: RF000 -> tpcds_15000_decimal_parquet.store_returns.sr_item_sk, RF003 -> sr_customer_sk, RF004 -> sr_item_sk, RF005 -> sr_returned_date_sk
      |     table stats: 4317083368 rows total
      |     column stats: all
      |     hosts=20 per-host-mem=352.00MB
      |     tuple-ids=1 row-size=36B cardinality=4317083368
      |
      18:EXCHANGE [HASH(ss_customer_sk,ss_item_sk,ss_ticket_number)]
      |  hosts=20 per-host-mem=0B
      |  tuple-ids=0,3 row-size=60B cardinality=3834191129
      |
      08:HASH JOIN [INNER JOIN, BROADCAST]
      |  hash predicates: ss_sold_date_sk = d1.d_date_sk
      |  runtime filters: RF009 <- d1.d_date_sk
      |  hosts=20 per-host-mem=3.11KB
      |  tuple-ids=0,3 row-size=60B cardinality=3834191129
      |
      |--17:EXCHANGE [BROADCAST]
      |  |  hosts=1 per-host-mem=0B
      |  |  tuple-ids=3 row-size=16B cardinality=181
      |  |
      |  03:SCAN HDFS [tpcds_15000_decimal_parquet.date_dim d1, RANDOM]
      |     partitions=1/1 files=1 size=2.17MB
      |     predicates: d1.d_year = 1999, d1.d_moy = 4
      |     table stats: 73049 rows total
      |     column stats: all
      |     hosts=1 per-host-mem=48.00MB
      |     tuple-ids=3 row-size=16B cardinality=181
      |
      00:SCAN HDFS [tpcds_15000_decimal_parquet.store_sales, RANDOM]
         partitions=1824/1824 files=6922 size=1.52TB
         runtime filters: RF000 -> ss_item_sk, RF001 -> ss_store_sk, RF003 -> tpcds_15000_decimal_parquet.store_sales.ss_customer_sk, RF004 -> tpcds_15000_decimal_parquet.store_sales.ss_item_sk, RF006 -> ss_customer_sk, RF007 -> ss_item_sk, RF008 -> ss_ticket_number, RF009 -> ss_sold_date_sk
         table stats: 43202446268 rows total
         column stats: all
         hosts=20 per-host-mem=440.00MB
         tuple-ids=0 row-size=44B cardinality=43202446268
      

      Summary

      Operator              #Hosts   Avg Time   Max Time    #Rows  Est. #Rows   Peak Mem  Est. Peak Mem  Detail                         
      ----------------------------------------------------------------------------------------------------------------------------------
      28:MERGING-EXCHANGE        1    1.109ms    1.109ms    1.02K         100          0        -1.00 B  UNPARTITIONED                  
      16:TOP-N                  20    3.652ms    5.942ms    2.00K         100  132.00 KB       21.89 KB                                 
      27:AGGREGATE              20  275.802ms  458.876ms  442.90K     335.85M   11.35 MB      212.22 MB  FINALIZE                       
      26:EXCHANGE               20   59.647ms  258.173ms    1.13M     335.85M          0              0  HASH(i_item_id,i_item_desc,... 
      15:AGGREGATE              20  343.414ms  422.845ms    1.13M     335.85M  143.24 MB       77.13 GB  STREAMING                      
      14:HASH JOIN              20   30.930ms   41.410ms    2.20M     335.85M   12.99 MB        5.24 MB  INNER JOIN, BROADCAST          
      |--25:EXCHANGE            20    1.052ms    1.677ms   32.00K      32.00K          0              0  BROADCAST                      
      |  07:SCAN HDFS            1   18.325ms   18.325ms   32.00K      32.00K    9.43 MB       48.00 MB  tpcds_15000_decimal_parquet... 
      13:HASH JOIN              20   18.778ms   25.275ms    2.20M     335.85M    3.49 MB        4.00 KB  INNER JOIN, BROADCAST          
      |--24:EXCHANGE            20   25.213us  301.428us       62          62          0              0  BROADCAST                      
      |  06:SCAN HDFS            1    5.887ms    5.887ms       62          62  159.00 KB       48.00 MB  tpcds_15000_decimal_parquet... 
      12:HASH JOIN              20   24.246ms   33.507ms    2.23M     335.85M    2.81 MB       25.23 KB  INNER JOIN, BROADCAST          
      |--23:EXCHANGE            20   34.326us   49.903us    1.10K       1.96K          0              0  BROADCAST                      
      |  05:SCAN HDFS            1   24.594ms   24.594ms    1.10K       1.96K    2.29 MB       32.00 MB  tpcds_15000_decimal_parquet... 
      11:HASH JOIN              20       4m8s      5m31s    2.23M     352.50M   38.14 GB       30.98 GB  INNER JOIN, PARTITIONED        
      |--22:EXCHANGE            20   20s051ms   54s305ms   12.85B      21.60B          0              0  HASH(cs_bill_customer_sk,cs... 
      |  02:SCAN HDFS           20    3s716ms    4s533ms   12.85B      21.60B  257.30 MB      264.00 MB  tpcds_15000_decimal_parquet... 
      21:EXCHANGE               20   64.831ms  101.634ms   16.49M     352.50M          0              0  HASH(sr_customer_sk,sr_item... 
      10:HASH JOIN              20  300.445ms  389.672ms   16.49M     352.50M    2.94 MB        3.54 KB  INNER JOIN, BROADCAST          
      |--20:EXCHANGE            20   11.347us   17.878us      122         206          0              0  BROADCAST                      
      |  04:SCAN HDFS            1   12.903ms   12.903ms      122         206    3.50 MB       48.00 MB  tpcds_15000_decimal_parquet... 
      09:HASH JOIN              20    8s854ms   11s946ms   16.49M       3.83B    1.01 GB        7.96 GB  INNER JOIN, PARTITIONED        
      |--19:EXCHANGE            20  444.716ms    1s130ms  234.43M       4.32B          0              0  HASH(sr_customer_sk,sr_item... 
      |  01:SCAN HDFS           20    1s235ms    1s397ms  234.43M       4.32B  351.41 MB      352.00 MB  tpcds_15000_decimal_parquet... 
      18:EXCHANGE               20    1s216ms    2s815ms  389.28M       3.83B          0              0  HASH(ss_customer_sk,ss_item... 
      08:HASH JOIN              20  949.230ms    2s216ms  389.28M       3.83B    2.71 MB        3.11 KB  INNER JOIN, BROADCAST          
      |--17:EXCHANGE            20   11.615us   16.053us       30         181          0              0  BROADCAST                      
      |  03:SCAN HDFS            1   14.187ms   14.187ms       30         181    2.89 MB       48.00 MB  tpcds_15000_decimal_parquet... 
      00:SCAN HDFS              20  620.986ms  759.453ms  389.28M      43.20B  344.09 MB      440.00 MB  tpcds_15000_decimal_parquet... 
      

        Attachments

          Issue Links

            Activity

              People

              • Assignee:
                Unassigned
                Reporter:
                mmokhtar Mostafa Mokhtar
              • Votes:
                0 Vote for this issue
                Watchers:
                5 Start watching this issue

                Dates

                • Created:
                  Updated: