Poor query plan based on values

aaron.pfeifer · December 1, 2018, 1:49am

Hi -

I appear to be running into an issue where the plan for a query against one of my columnstore tables in optimized significantly differently based on the values in a WHERE condition. In one case, a query is made against the table which matches zero rows. In another case, a query is made that matches multiple rows. Depending on which query is run first dramatically changes the query plan and, as a result, the performance.

For example, if I run the zero-row query first, the compiled query time is 0.01s for the zero-row query and 2s+ for the multi-row query.

If I run the multi-row query first, the compiled query time is 0.01s for both it and the zero-row query.

The explain for the zero-row query is:

| Project [GROUP_CONCAT(facts_merged.value SEPARATOR ','), GROUP_CONCAT(facts_merged.epoch SEPARATOR ','), facts_merged.metric]                                                                                                                                                 |
| HashGroupBy [GROUP_CONCAT(facts_merged.value SEPARATOR ',') AS values, GROUP_CONCAT(facts_merged.epoch SEPARATOR ',') AS epochs] groups:[facts_merged.metric]                                                                                                                 |
| TableScan 2tmp AS facts_merged storage:list stream:yes                                                                                                                                                                                                                        |
| Project [value, facts.epoch, facts.metric]                                                                                                                                                                                                                                    |
| HashGroupBy [SUM(facts.value) AS value] groups:[facts.epoch, facts.metric]                                                                                                                                                                                                    |
| TableScan 1tmp AS facts storage:list stream:yes                                                                                                                                                                                                                               |
| Project [value, remote_0.epoch, remote_0.metric]                                                                                                                                                                                                                              |
| HashGroupBy [SUM(remote_0.value) AS value] groups:[remote_0.epoch, remote_0.metric]                                                                                                                                                                                           |
| Gather partitions:all alias:remote_0                                                                                                                                                                                                                                          |
| Project [value, f.epoch, dim_metrics_ref.metric]                                                                                                                                                                                                                              |
| HashGroupBy [SUM(f.value) AS value] groups:[f.epoch, dim_metrics_ref.metric]                                                                                                                                                                                                  |
| Filter [dim_offers_ref.offer_id IN (...)]                                                                                                                                                                                                                                     |
| NestedLoopJoin                                                                                                                                                                                                                                                                |
| |---IndexSeek report_service.dim_offers_ref, PRIMARY KEY (id) scan:[id = f.offer_id] est_table_rows:182,510 est_filtered:2                                                                                                                                                    |
| Filter [dim_metrics_ref.metric IN (...)]                                                                                                                                                                                                                                      |
| NestedLoopJoin                                                                                                                                                                                                                                                                |
| |---IndexSeek report_service.dim_metrics_ref, PRIMARY KEY (id) scan:[id = f.metric_id] est_table_rows:138 est_filtered:35                                                                                                                                                     |
| Filter [f.epoch IN (...)]                                                                                                                                                                                                                                                     |
| ColumnStoreScan report_service.facts_offer_metrics_staging_rewrite AS f, KEY epoch (epoch, offer_id, metric_id, device_country_id, creative_id, app_id, store_id) USING CLUSTERED COLUMNSTORE WITH(COLUMNSTORE_SEGMENT_ROWS=1024000) est_table_rows:31,414,353 est_filtered:1 |

The explain for the multi-row query is:

| Project [GROUP_CONCAT(facts_merged.value SEPARATOR ','), GROUP_CONCAT(facts_merged.epoch SEPARATOR ','), facts_merged.metric]                                                                                                                                                          |
| HashGroupBy [GROUP_CONCAT(facts_merged.value SEPARATOR ',') AS values, GROUP_CONCAT(facts_merged.epoch SEPARATOR ',') AS epochs] groups:[facts_merged.metric]                                                                                                                          |
| TableScan 2tmp AS facts_merged storage:list stream:yes                                                                                                                                                                                                                                 |
| Project [value, facts.epoch, facts.metric]                                                                                                                                                                                                                                             |
| HashGroupBy [SUM(facts.value) AS value] groups:[facts.epoch, facts.metric]                                                                                                                                                                                                             |
| TableScan 1tmp AS facts storage:list stream:yes                                                                                                                                                                                                                                        |
| Project [value, remote_0.epoch, remote_0.metric]                                                                                                                                                                                                                                       |
| HashGroupBy [SUM(remote_0.value) AS value] groups:[remote_0.epoch, remote_0.metric]                                                                                                                                                                                                    |
| Gather partitions:all alias:remote_0                                                                                                                                                                                                                                                   |
| Project [value, f.epoch, dim_metrics_ref.metric]                                                                                                                                                                                                                                       |
| HashGroupBy [SUM(f.value) AS value] groups:[f.epoch, dim_metrics_ref.metric]                                                                                                                                                                                                           |
| HashJoin [f.metric_id = dim_metrics_ref.id]                                                                                                                                                                                                                                            |
| |---Project [dim_metrics_ref_0.metric, dim_metrics_ref_0.id] est_rows:35 alias:dim_metrics_ref                                                                                                                                                                                         |
| |   IndexSeek report_service.dim_metrics_ref AS dim_metrics_ref_0, UNIQUE KEY dim_metrics_metric_key (metric) scan:[metric IN (...)] est_table_rows:138 est_filtered:35                                                                                                                |
| HashJoin [f.offer_id = dim_offers_ref.id]                                                                                                                                                                                                                                              |
| |---Project [dim_offers_ref_0.id] est_rows:1 alias:dim_offers_ref                                                                                                                                                                                                                      |
| |   IndexSeek report_service.dim_offers_ref AS dim_offers_ref_0, UNIQUE KEY dim_offers_offer_id_key (offer_id) scan:[offer_id IN (...)] est_table_rows:182,510 est_filtered:2                                                                                                          |
| Filter [f.epoch IN (...)]                                                                                                                                                                                                                                                              |
| BloomFilter table:dim_metrics_ref fields:f.metric_id                                                                                                                                                                                                                                   |
| BloomFilter table:dim_offers_ref fields:f.offer_id                                                                                                                                                                                                                                     |
| ColumnStoreScan report_service.facts_offer_metrics_staging_rewrite AS f, KEY epoch (epoch, offer_id, metric_id, device_country_id, creative_id, app_id, store_id) USING CLUSTERED COLUMNSTORE WITH(COLUMNSTORE_SEGMENT_ROWS=1024000) est_table_rows:31,414,353 est_filtered:31,414,353 |

I’ve tried rewriting the query in various ways, but haven’t been able to force the query optimized to use the correct plan for the zero-row query. I’m not quite sure how I can guarantee that the right plan gets used since this is controlled by the order in which queries come into the database.

I’d love any thoughts you might have. Thanks!

aaron.pfeifer · December 1, 2018, 1:52am

Also FYI I’m running this against MemSQL 6.5.8.

rob · December 1, 2018, 2:56am

Query plans are currently optimized the first time they are run and then stored in the plancache for subsequent executions, with parameters removed. This works great if the selectivity of the parameters used in filters is roughly similar across multiple executions of the query but not if different parameters would generate very different plans. In the future, we are planning on adding a mechanism to only share the plan if the filter selectivity would be similar and generate equivalent plans.

For now, the best way to workaround this would be to manually hint the query plan that works well for both cases. If you share the SQL text, I can share the hints that would improve this query.

aaron.pfeifer · December 1, 2018, 3:49am

Thanks for the follow up! That makes sense.

Our current query is structured like so:

select
  sum(value) as value,
  f.epoch as epoch,
  dim_metrics.metric as metric
from
  facts f
  inner join dim_metrics on dim_metrics.id = f.metric_id
  inner join dim_offers on dim_offers.id = f.offer_id
where
  dim_metrics.metric in (...)
  and f.epoch between 1543521600 and 1543604400
  and dim_offers.offer_id in (...)
group by
  f.epoch,
  dim_metrics.metric;

If there’s a way to provide a hint, that’d be great.

rob · December 1, 2018, 9:13pm

To get the right plan for the query, we just need to force the hash join. The hint with(table_convert_subselect=true) applied to each table we want to do a hash join with would be sufficient here. This hint will add a scan to any table it’s applied to and the optimizer will pick hash join as a result since the indexes will not be available to do a nested loop index join.

select
  sum(value) as value,
  f.epoch as epoch,
  dim_metrics.metric as metric
from
  facts f
  inner join dim_metrics with(table_convert_subselect=true) on dim_metrics.id = f.metric_id
  inner join dim_offers with(table_convert_subselect=true) on dim_offers.id = f.offer_id
where
  dim_metrics.metric in (...)
  and f.epoch between 1543521600 and 1543604400
  and dim_offers.offer_id in (...)
group by
  f.epoch,
  dim_metrics.metric;

aaron.pfeifer · December 3, 2018, 2:15am

Hi rob -

Thanks for the suggestion! That did the trick! Is there a reference for the various optimizer hints that are available in MemSQL? I didn’t notice any documentation on that particular option.

-Aaron

aaron.pfeifer · December 4, 2018, 4:50pm

Hey rob -

After further testing, we’ve found that in some cases adding the with(table_convert_subselect=true) still results in poor performance due to the values being used in the filters. In fact, we’ve found that we need to only add a flag to one of our join tables in order for the better query plan to kick in.

For example, this is our original query:

select
  sum(value) as value,
  epoch as epoch,
  dim_metrics_ref.metric as metric
from
  facts_offer_metrics f
  inner join dim_metrics_ref on dim_metrics_ref.id = f.metric_id
  inner join dim_offers_ref on dim_offers_ref.id = f.offer_id
where
  dim_metrics_ref.metric in (...)
  and epoch in (...)
  and dim_offers_ref.offer_id in (...)
group by
  epoch,
  dim_metrics_ref.metric;

Depending on which values we pass in to some of the where condition filters, the performance ends up being significantly worse due to the way the query plan gets built for one query vs. the other. So – depending on which query gets run first, we may or may not get the most efficient plan.

When we add with(table_convert_subselect=true) to both join tables, the second query is still slow. If we add the flag to only the dim_offers_ref join table, the query becomes fast.

Here are the explains on the query when the flag is (a) not present, (b), present on both joins, and © present on just the dim_offers_ref join table.

A. No flag

+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| EXPLAIN                                                                                                                                                                                                                                                          |
+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| Project [value, remote_0.epoch, remote_0.metric]                                                                                                                                                                                                                 |
| HashGroupBy [SUM(remote_0.value) AS value] groups:[remote_0.epoch, remote_0.metric]                                                                                                                                                                              |
| Gather partitions:all alias:remote_0                                                                                                                                                                                                                             |
| Project [value, f.epoch, dim_metrics_ref.metric]                                                                                                                                                                                                                 |
| HashGroupBy [SUM(f.value) AS value] groups:[f.epoch, dim_metrics_ref.metric]                                                                                                                                                                                     |
| Filter [dim_offers_ref.offer_id IN (...)]                                                                                                                                                                                                                        |
| NestedLoopJoin                                                                                                                                                                                                                                                   |
| |---IndexSeek report_service.dim_offers_ref, PRIMARY KEY (id) scan:[id = f.offer_id] est_table_rows:183,368 est_filtered:1                                                                                                                                       |
| Filter [dim_metrics_ref.metric IN (...)]                                                                                                                                                                                                                         |
| NestedLoopJoin                                                                                                                                                                                                                                                   |
| |---IndexSeek report_service.dim_metrics_ref, PRIMARY KEY (id) scan:[id = f.metric_id] est_table_rows:138 est_filtered:35                                                                                                                                        |
| Filter [f.epoch IN (...)]                                                                                                                                                                                                                                        |
| ColumnStoreScan report_service.facts_offer_metrics AS f, KEY epoch (epoch, offer_id, metric_id, device_country_id, creative_id, app_id, store_id) USING CLUSTERED COLUMNSTORE WITH(COLUMNSTORE_SEGMENT_ROWS=1024000) est_table_rows:5,821,315,105 est_filtered:1 |
+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+

B. Flag on both joins

+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| EXPLAIN                                                                                                                                                                                                                                                                |
+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| Project [value, remote_0.epoch, remote_0.metric]                                                                                                                                                                                                                       |
| HashGroupBy [SUM(remote_0.value) AS value] groups:[remote_0.epoch, remote_0.metric]                                                                                                                                                                                    |
| Gather partitions:all alias:remote_0                                                                                                                                                                                                                                   |
| Project [value, f.epoch, dim_metrics_ref.metric]                                                                                                                                                                                                                       |
| HashGroupBy [SUM(f.value) AS value] groups:[f.epoch, dim_metrics_ref.metric]                                                                                                                                                                                           |
| HashJoin [f.offer_id = dim_offers_ref.id]                                                                                                                                                                                                                              |
| |---Project [dim_offers_ref_0.id] est_rows:1 alias:dim_offers_ref                                                                                                                                                                                                      |
| |   IndexSeek report_service.dim_offers_ref AS dim_offers_ref_0, UNIQUE KEY dim_offers_offer_id_key (offer_id) scan:[offer_id IN (...)] est_table_rows:183,368 est_filtered:1                                                                                          |
| BloomFilter table:dim_offers_ref fields:f.offer_id                                                                                                                                                                                                                     |
| HashJoin [f.metric_id = dim_metrics_ref.id]                                                                                                                                                                                                                            |
| |---Project [f_0.value, f_0.epoch, f_0.offer_id, f_0.metric_id] est_rows:48 alias:f                                                                                                                                                                                    |
| |   Filter [f_0.epoch IN (...)]                                                                                                                                                                                                                                        |
| |   ColumnStoreScan report_service.facts_offer_metrics AS f_0, KEY epoch (epoch, offer_id, metric_id, device_country_id, creative_id, app_id, store_id) USING CLUSTERED COLUMNSTORE WITH(COLUMNSTORE_SEGMENT_ROWS=1024000) est_table_rows:5,821,315,105 est_filtered:1 |
| TableScan 0tmp AS dim_metrics_ref storage:list stream:yes est_table_rows:35                                                                                                                                                                                            |
| Project [dim_metrics_ref_0.metric, dim_metrics_ref_0.id] est_rows:35                                                                                                                                                                                                   |
| IndexSeek report_service.dim_metrics_ref AS dim_metrics_ref_0, UNIQUE KEY dim_metrics_metric_key (metric) scan:[metric IN (...)] est_table_rows:138 est_filtered:35                                                                                                    |
+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+

C. Flag on just dim_offers_ref

+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| EXPLAIN                                                                                                                                                                                                                                                          |
+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| Project [value, remote_0.epoch, remote_0.metric]                                                                                                                                                                                                                 |
| HashGroupBy [SUM(remote_0.value) AS value] groups:[remote_0.epoch, remote_0.metric]                                                                                                                                                                              |
| Gather partitions:all alias:remote_0                                                                                                                                                                                                                             |
| Project [value, f.epoch, dim_metrics_ref.metric]                                                                                                                                                                                                                 |
| HashGroupBy [SUM(f.value) AS value] groups:[f.epoch, dim_metrics_ref.metric]                                                                                                                                                                                     |
| HashJoin [f.offer_id = dim_offers_ref.id]                                                                                                                                                                                                                        |
| |---Project [dim_offers_ref_0.id] est_rows:1 alias:dim_offers_ref                                                                                                                                                                                                |
| |   IndexSeek report_service.dim_offers_ref AS dim_offers_ref_0, UNIQUE KEY dim_offers_offer_id_key (offer_id) scan:[offer_id IN (...)] est_table_rows:183,368 est_filtered:1                                                                                    |
| Filter [dim_metrics_ref.metric IN (...)]                                                                                                                                                                                                                         |
| NestedLoopJoin                                                                                                                                                                                                                                                   |
| |---IndexSeek report_service.dim_metrics_ref, PRIMARY KEY (id) scan:[id = f.metric_id] est_table_rows:138 est_filtered:35                                                                                                                                        |
| Filter [f.epoch IN (...)]                                                                                                                                                                                                                                        |
| BloomFilter table:dim_offers_ref fields:f.offer_id                                                                                                                                                                                                               |
| ColumnStoreScan report_service.facts_offer_metrics AS f, KEY epoch (epoch, offer_id, metric_id, device_country_id, creative_id, app_id, store_id) USING CLUSTERED COLUMNSTORE WITH(COLUMNSTORE_SEGMENT_ROWS=1024000) est_table_rows:5,821,315,105 est_filtered:1 |
+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+

As you can see, we start to have this tricky decision of knowing which join table we should add the flag to. We can have up to 6 or 7 join tables and it all depends on user input coming in to our service.

Any other guidance you might have to implement a general way we can build our queries to know which join tables to add the flag to?

aaron.pfeifer · December 4, 2018, 5:15pm

As another data point to help provide context, if we switch over to explicit foreign key filters instead of joining to get the foreign keys, we don’t seem to need the optimizer hints at all. This is the query plan we see:

Query:

select
  sum(value) as value,
  epoch as epoch,
  dim_metrics_ref.metric as metric
from
  facts_offer_metrics f
  inner join dim_metrics_ref on dim_metrics_ref.id = f.metric_id
where
  f.metric_id in (...)
  and epoch in (...)
  and f.offer_id in (...)
group by
  epoch,
  dim_metrics_ref.metric;

Plan:

+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| EXPLAIN                                                                                                                                                                                                                                                          |
+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| Project [value, remote_0.epoch, remote_0.metric]                                                                                                                                                                                                                 |
| HashGroupBy [SUM(remote_0.value) AS value] groups:[remote_0.epoch, remote_0.metric]                                                                                                                                                                              |
| Gather partitions:all alias:remote_0                                                                                                                                                                                                                             |
| Project [value, f.epoch, dim_metrics_ref.metric]                                                                                                                                                                                                                 |
| HashGroupBy [SUM(f.value) AS value] groups:[f.epoch, dim_metrics_ref.metric]                                                                                                                                                                                     |
| Filter [dim_metrics_ref.id IN (...)]                                                                                                                                                                                                                             |
| NestedLoopJoin                                                                                                                                                                                                                                                   |
| |---IndexSeek report_service.dim_metrics_ref, PRIMARY KEY (id) scan:[id = f.metric_id] est_table_rows:138 est_filtered:35                                                                                                                                        |
| Filter [f.metric_id IN (...) AND f.epoch IN (...) AND f.offer_id IN (...)]                                                                                                                                                                                       |
| ColumnStoreScan report_service.facts_offer_metrics AS f, KEY epoch (epoch, offer_id, metric_id, device_country_id, creative_id, app_id, store_id) USING CLUSTERED COLUMNSTORE WITH(COLUMNSTORE_SEGMENT_ROWS=1024000) est_table_rows:5,826,473,568 est_filtered:1 |
+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+

That definitely has the best performance from any of the other queries. I think I had mentioned it in another ticket, but obviously this creates more work on our client-side which we’d ideally like to avoid.

rob · December 6, 2018, 5:27pm

In general, the plan issues are due to the varying selectivity of the filters in your queries and misestimations of them. One solution to try to this would be to hint the filter selectivity on the facts table so that it always generates the plan we want. We can hint this via with(selectivity=...).

select
  sum(value) as value,
  f.epoch as epoch,
  dim_metrics.metric as metric
from
  facts f with(selectivity=1.0)
  join dim_metrics on dim_metrics.id = f.metric_id
  join dim_offers on dim_offers.id = f.offer_id
where
  dim_metrics.metric in (...)
  and f.epoch between 1543521600 and 1543604400
  and dim_offers.offer_id in (...)
group by
  f.epoch,
  dim_metrics.metric;

Note that it’s not particularly important that the selectivity we hint is accurate, just that it will generate the fast plan we want, which is fact first followed by hash joins with any of the dim tables.

There were some estimation improvements in 6.7 for IN filters (they can use histograms now), so would be interesting to see how these queries get estimated there.