Join Reordering and Join Distribution Type Determination Based on Table Size

Presto version 0.208 on Qubole now has the ability to do stats-based determination of JOIN-distribution-type (between BROADCAST and PARTITIONED); back ported from the Presto open source version 0.210. It means that if table statistics are present, Presto’s cost-based-optimizer tries to optimise a query plan by choosing the right type of JOIN implementation on the basis of memory, CPU and network cost for every JOIN node in the plan.

Note

This feature is available for a beta access and it is only available for Hive tables.

For using the automatic JOIN type determination, you can set:

  • join-distribution-type=AUTOMATIC in config.properties at the cluster level
  • join_distribution_type=AUTOMATIC in session properties

Note

This feature cannot be used if the property distributed-join is already set in the session or cluster config.properties.

This feature requires table statistics to be present for the cost based optimiser to determine join distribution type. Further, the statistics should be maintained and updated as the table evolves for correct estimates. To address the collection and maintenance of the statistics, Qubole provides an Automatic Stats Collection framework.

Further, Qubole has also introduced the notion of estimating table statistics on the basis of the table’s size on the storage layer. It is useful in cases where statistics for tables are not available. If Presto on Qubole is unable to find table statistics, it can fetch the size of the table on the storage layer and estimate the size and the number of rows in the table. This estimate can currently be used to determine the JOIN distribution type and reordering of tables in a multi-JOIN scenario.

For enabling table size based stats to determine join distribution type, you can set:

  • At the cluster level under Advanced Configuration > PRESTO SETTINGS > Override Presto Configuration in the cluster’s UI.

    config.properties:
    join-distribution-type=AUTOMATIC
    enable-file-size-stats-join-type=true
    
  • In a query as session properties

    join-distribution-type=AUTOMATIC
    enable_file_size_stats_join_type=true
    

For enabling table size-based stats to determine JOIN order, you can set:

  • At the cluster level under Advanced Configuration > PRESTO SETTINGS > Override Presto Configuration in the cluster’s UI.

    config.properties:
    optimizer.join-reordering-strategy=AUTOMATIC
    join-distribution-type=AUTOMATIC
    enable-file-size-stats-join-reorder=true
    
  • In a query as session properties

    join_reordering_strategy=AUTOMATIC
    join_distribution_type=AUTOMATIC
    enable_file_size_stats_join_reorder=true
    

Configuring Timeout for Fetching Hive Table Size

hive.table-size-listing-timeout is the property that you can use to set the timeout for listing Hive table sizes. The determination of a table size on the storage layer is performed through a listing on the table’s storage location. The listing operation is bound by a timeout to avoid any significant delays in the query execution time. This configuration controls the timeout for the listing operation.

An example usage of this configuration is adding hive.table-size-listing-timeout=2s to the Hive catalog properties. It would mean that the listing operation on a table’s storage location is bound to complete in 2 seconds. If it does not finish within the timeout, the table is considered to be very large.