Presto Performance Tuning

Review the following information to learn how to tune Presto for the best performance. These performance tuning tips assume the following level of experience:

• Basic knowledge of Treasure Data.
• Basic knowledge of Presto query engine. Learn how to write Presto Queries .

Tuning Tips

Review these tips and make adjustments to increase your Presto queries performance.

• Always use TD_TIME_RANGE or TD_INTERVAL (with a link to Scheduling Presto Queries)
• Avoid using slow memory consuming operators such as
  • ORDER BY
  • COUNT(DISTINCT x)

For example, instead of using COUNT (DISTINCT x), use approx_distinct(x).

• Join processing
  • Tables should be joined in the order of a larger table to smaller tables.
  • Using a non-equi join condition slows down the query processing.
• Columnar storage characteristics
  • Choosing too many columns slows down query processing.
• Query result size
  • Generating too many rows takes time. Instead, use CREATE TABLE AS... or INSERT INTO or result_output_redirect magic comment .

Specifying the Columns you Need

Actual data in Treasure Data is stored as a columnar storage format, which is optimized for the query using only specific columns. Restricting accessed columns can improve your query performance significantly. Specify only needed columns instead of using a wildcard (*).

[GOOD]: SELECT time,user,host FROM tbl
[BAD]:  SELECT * FROM tbl

Leveraging Time-Based Partitioning

All imported data is automatically partitioned into hourly buckets, based on the time field within each data record.

By specifying the time range in your query, you can avoid reading unnecessary data and can thus speed up your query significantly.

Specify Time as an Integer

When the time field is specified within a WHERE clause, the query parser automatically detects which partitions should be processed. This auto-detection does not work if you specify the time with float instead of int.

[GOOD]: SELECT field1, field2, field3 FROM tbl WHERE time > 1349393020
[GOOD]: SELECT field1, field2, field3 FROM tbl WHERE time > 1349393020 + 3600
[GOOD]: SELECT field1, field2, field3 FROM tbl WHERE time > 1349393020 - 3600
[GOOD]: SELECT field1, field2, field3 FROM tbl WHEREtime > ${last_proceessed_time} 
[BAD]:  SELECT field1, field2, field3 FROM tbl WHERE time > 13493930200 / 10
[BAD]:  SELECT field1, field2, field3 FROM tbl WHERE time > 1349393020.00
[BAD]:  SELECT field1, field2, field3 FROM tbl WHERE time BETWEEN 1349392000 AND 1349394000
[BAD]:  SELECT field1, field2, field3 FROM tbl WHEREtime > (SELECT MAX(last_updated) FROM tbl2) 

Use TD-TIME-RANGE

You can use TD_TIME_RANGE to partition data.

[GOOD]: SELECT ... WHERE TD_TIME_RANGE(time, '2013-01-01 PDT')
[GOOD]: SELECT ... WHERE TD_TIME_RANGE(time, '2013-01-01','PDT', NULL)
[GOOD]: SELECT ... WHERE TD_TIME_RANGE(time, '2013-01-01',
                                       TD_TIME_ADD('2013-01-01', '1day', 'PDT'))

xHowever, if you use division in TD_TIME_RANGE, the time partition optimization doesn’t work. For example, Treasure Data discourages the SQL constructions as they disable optimization.

Poorly optimized SQL

SELECT ... WHERE TD_TIME_RANGE(time, TD_SCHEDULED_TIME() / 86400 * 86400))

Poorly optimized SQL

SELECT ... WHERE TD_TIME_RANGE(time, 1356998401 / 86400 * 86400))

The TD_INTERVAL user defined function is also available for partitioning. TD_INTERVAL provides an intuitive way to specify the time range. For example, to select the last 7 days:

SELECT ... WHERE TD_INTERVAL(time, '-7d')

Considering the Cardinality within the GROUP BY Clause

It's possible to improve the performance of the GROUP BY function by carefully ordering a list of fields within the GROUP BY in an order of high cardinality.

Or, use numbers instead of strings for the GROUP BY column, because numbers require less memory and are faster to compare than strings.

Using LIMIT with ORDER BY

ORDER BY requires that all rows be sent to a single worker which then sorts them. ORDER BY might often require a lot of memory on a Presto worker. When you want to look up the top or bottom N records, use LIMIT which can reduce the sort cost and memory pressure.

Using Approximate Aggregate Functions

Presto has a couple of approximate aggregation functions, that give you significant performance improvements, but with some errors. For example, by using approx_distinct() function, you can get an approximation of COUNT(DISTINCT x)with a standard error of 2.3%. The following example gives an approximate count of the previous day’s unique users.

SELECT
  approx_distinct(user_id)
FROM
  access
WHERE
  TD_TIME_RANGE(time,
    TD_TIME_ADD(TD_SCHEDULED_TIME(), '-1d', 'PDT'),
    TD_SCHEDULED_TIME())

Aggregating a Series of LIKE Clauses in One Single regexp_like Clause

Presto’s query optimizer is unable to improve queries that contain many LIKE clauses. As a consequence, the query execution might be slower than expected.

To improve the performance, you can substitute a series of LIKE clauses that are chained with the OR condition with a single regexp_like clause, which is Presto native.

For example:

SELECT
  ...
FROM
  access
WHERE
  method LIKE '%GET%' OR
  method LIKE '%POST%' OR
  method LIKE '%PUT%' OR
  method LIKE '%DELETE%'

can be optimized by replacing the four LIKE clauses with a single regexp_like clause:

SELECT
  ...
FROM
  access
WHERE
  regexp_like(method, 'GET|POST|PUT|DELETE')

Choosing Presto Join and Sort Algorithms

Join Algorithms

There are two types of join distributions:

• Partitioned: each node participating in the query builds a hash table from a fraction of the data
• Broadcast: each node participating in the query builds a hash table from all of the data (data is replicated to each node)

The default join algorithm used in Presto is distributed PARTITION join. This algorithm partitions both the left and right tables using the hash values of join keys. Partitioned join uses multiple worker nodes for processing queries; it is generally faster and takes less memory.

In some cases where one of the join tables is very small, the overhead of partitioning data through the network using distributed PARTITION join might exceed the benefit of broadcasting the whole table to all the nodes participating in the join operation. In those cases, 'BROADCAST' join might perform better. If you use the BROADCAST join, specify the large table first in the join clause. 'BROADCAST' join can be enabled specifying the following magic comment.

-- set session join_distribution_type = 'BROADCAST'

This option will use more memory as the right join table will be copied to all nodes.

Sort Algorithms

For queries with ORDER BY clause, Presto uses Distributed sort by default; the sorting operation runs in parallel on multiple nodes, and a single node merges the final results. However, there might be cases where a single node sort operation performs better. For such cases, the distributed sort can be disabled by using the following magic comment.

-- set distributed_sort=false

Disabling distributed sort will use more memory of a single node; the query might fail due to the maximum memory limit.

Defining Partitioning for Presto

All tables in Treasure Data are partitioned based on the time column. This is why queries that use TD_TIME_RANGE or similar predicates on the time column are efficient in Treasure Data. Presto can eliminate partitions that fall outside the specified time range without reading them.

User-defined partitioning (UDP) provides hash partitioning for a table on one or more columns in addition to the time column. A query that filters on the set of columns used as user-defined partitioning keys can be more efficient because Presto can skip scanning partitions that have matching values on that set of columns. Use UDP when fast ID lookup is necessary.

Managing Resources for Presto Queries

• Use resource pools
• Run more concurrent queries
• Use more CPU resources
• Use more memory (limited)

Magic Comments

Magic comments are instructions inserted into Presto SQL queries and given by the database driver to override the database optimization that occurs with standard Presto queries. The database driver follows certain rules for adding the magic comments to the database query and the magic comments override any execution plan the query optimizer might select for a query.

Use magic comments in cases when you know more about the database content and structure than the query optimizer. It can be more efficient to use magic comments to improve execution time.

Magic Comment Parsing Decimal Literals as Double

A Presto session-level property that controls whether the decimal literals (ex. 1.2) are parsed as decimal or double.

The parse_decimal_literals_as_double session property is supported as a magic comment.

Decimal literals without an explicit type specifier (e.g. 1.2) are treated as the values of the DOUBLE type by default.

This magic comment can disable the behavior and use decimal literals as the values of the DECIMAL type. For example:

-- set session parse_decimal_literals_as_double = 'false'
SELECT * FROM large_table, small_table
WHERE small_table.id = large_table.id

Magic Comment Distributed Sort

Presto session property that when enabled allows the sort operator to execute in parallel on multiple nodes in the cluster.

Distributed sort allows sorting of data, which exceeds query.max-memory-per-node. Distributed sort is enabled through the distributed_sort session property, or distributed-sort configuration property set in etc/config.properties of the coordinator. Distributed sort is enabled by default.

When distributed sort is enabled, the sort operator executes in parallel on multiple nodes in the cluster. Partially sorted data from each worker node is then streamed to a single worker node for a final merge. This technique allows to utilization of memory of multiple worker nodes for sorting. The primary purpose of the distributed sort is to allow for the sorting of data sets that don’t normally fit into single node memory. Performance improvement can be expected, but it won’t scale linearly with the number of nodes, since the data needs to be merged by a single node.

-- set session distributed_sort = 'true'
SELECT * FROM large_table, small_table
WHERE small_table.id = large_table.id

Magic Comment Time Partitioning Range

A TD-specific setting to customize partition size for INSERT/CTAS queries.

The timepartitioningrange session property is supported as a magic comment.

-- set session time_partitioning_range = 'value'

The value can be:

• none
  • no time partitioning
• number with unit
  • allowed units are h, d, mo, q, y (hour, day, month, quarter, year). For example, 12h, 2d, 1mo, 1q, 1y

Magic Comment Limiting Execution Time

If you want to override any other limits for a single query, you can apply the following query hint at the beginning of the query:

-- set session query_max_execution_time = '2h' 
SELECT 
  COUNT(*)
FROM
  ALL_EVENTS
;

If limits are set in your account, and you write a query with a hint, then the smallest limit between account, resource-group, query-hint, and global is used. Supported time measures are h (hours), m (minutes), s (seconds).

If the limit specified in the query hint is longer than the configured limit, then the shorter configured limit still applies. For example, if a 1-hour limit is configured for ad hoc queries, then you can use the magic comment to limit a query to 5 minutes (5m) but not to increase the limit to 2 hours (2h).

Magic Comment Extend Query Runtime Beyond Configured Limits

You can run a query in a special overcommit resource pool if you want to extend the runtime beyond the query's normal limits. In this case, the limit specified in the query hint is greater than the query's normal limit.

For example, if there is a 30-minute limit configured on scheduled queries, you can run a longer query with the following query hints:

-- set property resource_group = 'overcommit'
-- set session query_max_execution_time = '1d'' 

Magic Comment to improve performance for result output

To improve the job performance of queries that produce massive results, result_output_redirect is used to enhance the performance of Presto jobs. By default, result_output_redirect is set to 'true'.

-- set session result_output_redirect='true'

This impacts performance even though the query results are ready; TD Presto is waiting for the worker node to complete its sequential operations. To mitigate this, Treasure Data now uses result_output_direct, which redirects the query result to S3 in parallel, thereby improving the performance of the queries.

LIMITATION: There are SQL queries where Hints do not improve performance, including:

• When a query includes ORDER BY
• When a query is CREATE TABLE or INSERT TABLE
• When a query includes uppercase letters
• When a query has duplicated column names
• When a query has NULL without explicit type

Limitations

The benefits of UDP can be limited when used with more complex queries. The query optimizer might not always apply UDP in cases where it can be beneficial.

The Maximum Number of Partitioning Keys is Less Than or Equal to Three Keys

If the limit is exceeded, Presto causes the following error message:

'bucketed_on' must be less than 4 columns

Imports Other than SQL

The import method provided by Treasure Data for the following does not support UDP tables:

• Streaming Import
• Data Connector
• BulkImport

If you try to use any of these import methods, you will get an error. As a workaround, you can use a workflow to copy data from a table that is receiving streaming imports to the UDP table.

Use Cases for UDP

Where the lookup and aggregations are based on one or more specific columns, UDP can lead to:

• efficient lookup and aggregation queries
• efficient join queries

UDP can add the most value when records are filtered or joined frequently by non-time attributes::

• a customer's ID, first name+last name+birth date, gender, or other profile values or flags
• a product's SKU number, bar code, manufacturer, or other exact-match attributes
• an address's country code; city, state, or province; or postal code

Performance benefits become more significant on tables with >100M rows.

UDP can help with these Presto query types:

• "Needle-in-a-Haystack" lookup on the partition key
• Aggregations on the partition key
• Very large joins on partition keys used in tables on both sides of the join

Basic UDP Usage

CREATE TABLE Syntax for UDP

To create a UDP table:

1. Use CREATE TABLE with the attributes bucketed _ on to identify the bucketing keys and bucket _ count for the number of buckets.
2. Optionally, define the max _ fil _ size and max _ time _ range values.
   Copy

   Copied

   CREATE
   TABLE
   table_name WITH(
     bucketed_on = array['col1' [,
     'col2']... ] [,
     bucket_count = n] [,
     max_file_size = 'nnnMB'] [,
     max_time_range = 'nnd']
   ) [AS SELECT 
     . . . ]
   ;

For bucket_count the default value is 512. This should work for most use cases.

For example:

CREATE
  TABLE mytable_p(
    time bigint,
    col1 VARCHAR,
    col2 VARCHAR,
    col3 bigint
  ) WITH (
    bucketed_on = array['col1' [,
    'col2']... ]
  )
;If not specified:

• max _ file _ size will default to 256MB partitions
• max _ time _ range to 1d or 24 hours for time partitioning
• bucket _ count = 512

Choosing Bucketing Columns for UDP

Supported TD data types for UDP partition keys include int, long, and string. These correspond to Presto data types as described in About TD Primitive Data Types.

Choose a set of one or more columns used widely to select data for analysis &mdash that is, one frequently used to look up results, drill down to details, or aggregate data. For example, depending on the most frequently used types, you might choose:

• customer_id
• Country + State/Province + City
• Postal_code
• Customer-first name + last name + date of birth

Choose a column or set of columns that have high cardinality (relative to the number of buckets), and are frequently used with equality predicates. For example:

• Unique values, for example, an email address or account number
• Non-unique but high-cardinality columns with relatively even distribution, for example, date of birth

Checking For and Addressing Data Skew

The performance is inconsistent if the number of rows in each bucket is not roughly equal. For example, if you partition on the US zip code, urban postal codes will have more customers than rural ones.

To help determine bucket count and partition size, you can run a SQL query that identifies distinct key column combinations and counts their occurrences. For example:

SELECT 
  (
    concat(key_column1 [,
      '|',
      key_column2])
  ),
  COUNT(*)
FROM
  tbl. . .
GROUP BY
  1
;

If the counts across different buckets are roughly comparable, your data is not skewed.

For consistent results, choose a combination of columns where the distribution is roughly equal.

If you do decide to use partitioning keys that do not produce an even distribution, see Improving Performance with Skewed Data.

Using INSERT and INSERT OVERWRITE to Partitioned Tables

INSERT and INSERT OVERWRITE with partitioned tables work the same as with other tables. You can create an empty UDP table and then insert data into it the usual way. The resulting data is partitioned.

Partitioning an Existing Table

Tables must have partitioning specified when first created. For an existing table, you must create a copy of the table with UDP options configured and copy the rows over. To do this use a CTAS from the source table.

For example:

drop table if EXISTS customer_p
;

create table customer_p with (
  bucketed_on = array['customer_id'],
  bucket_count = 512
)  as select  *  from  customer

When partitioning an existing table:

• Creating a partitioned version of a very large table is likely to take hours or days. Consult with TD support to make sure you can complete this operation.
• If the source table is continuing to receive updates, you must update it further with SQL. For example:

    INSERT
      INTO
        customer_p SELECT 
           *
        FROM
          customer WHERE... 
    ```

Creating and Using UDP Tables: Examples

Create a partitioned copy of the customer table named customer_p, to speed up lookups by customer_id;

drop table if EXISTS customer_p
;

create table customer_p with (
  bucketed_on = array['customer_id'],
  bucket_count = 512
)   as select  *  from  customer

Create and populate a partitioned table customers_p to speed up lookups on "city+state" columns:

-- create partitioned table, bucketing on combination of city + state columns
-- create table customer_p with (bucketed_on = array['city','state'] , bucket_count=512, max_file_size = '256MB', max_time_range='30d');
create table customer_p with (
  bucketed_on = array['city',
  'state'],
  bucket_count = 512
)
;

-- update for beta
-- Insert new records into the partitioned table 
INSERT
  INTO
    customer_p
  VALUES(
    . . . . .
  )
;

INSERT
  overwrite INTO
    customer_p
  VALUES(
    . . . .
  )
;

-- accelerates queries that test equality on all bucketing columns 
SELECT 
   *
FROM
  customer_p
WHERE
  city = 'San Francisco'
  AND state = 'California'
;

SELECT 
   *
FROM
  customer_p
WHERE
  city IN(
    'San Francisco',
    'San Jose'
  )
  AND state = 'California'
  AND monthly_income > 10000
;

-- NOT accelerated: filter predicate does not use all hash columns
SELECT 
   *
FROM
  customer_p
WHERE
  state = 'California'
;

SELECT 
   *
FROM
  customer_p
WHERE
  city IN(
    'San Francisco',
    'San Jose'
  )
; 

UDP Advanced Use Case Details

• Choosing Bucket Count, Partition Size in Storage, and Time Ranges for Partitions
• Aggregations on the Hash Key
• Needle-in-a-Haystack Lookup on the Hash Key
• Very Large Join Operations
• Improving Performance on Skewed Data

Choosing Bucket Count, Partition Size in Storage, and Time Ranges for Partitions

Bucket counts must be in powers of two. A higher bucket count means dividing data among many smaller partitions, which can be less efficient to scan. TD suggests starting with 512 for most cases. If you aren't sure of the best bucket count, it is safer to err on the low side.

We recommend partitioning UDP tables on one-day or multiple-day time ranges, instead of the one-hour partitions most commonly used in TD. Otherwise, you might incur higher costs and slower data access because too many small partitions have to be fetched from storage.

Aggregations on the Hash Key

Using a GROUP BY key as the bucketing key, major improvements in performance and reduction in cluster load on aggregation queries were seen. For example, you can see that the UDP version of this query on a 1TB table

• used 10 Presto workers instead of 19
• ran in 45 seconds instead of 2 minutes 31 seconds
• processed three times as many rows per second.

The total data processed in GB was greater because the UDP version of the table occupied more storage.

UDP Version

presto: udp_tpch_sf1000 > SELECT 
  COUNT(*)
FROM
  (
    SELECT 
      max_by(
        l_discount,
        time
      ),
      max_by(
        l_partkey,
        time
      )
    FROM
      lineitem
    GROUP BY
      l_orderkey
  )
;

_col0------------
1500000000(
  1 row
) Query 20171227_014452_14154_sjh9g,
FINISHED,
10 nodes Splits: 517 total,
517 done(
  100.00 %
) 0: 45 [6B ROWS,
25.5GB] [134M ROWS / s,
585MB / s]

Non-UDP Version

presto: udp_tpch_sf1000 > SELECT 
  COUNT(*)
FROM
  (
    SELECT 
      max_by(
        l_discount,
        time
      ),
      max_by(
        l_partkey,
        time
      )
    FROM
      tpch_sf1000.lineitem
    GROUP BY
      l_orderkey
  )
;

_col0------------
1500000000(
  1 row
) Query 20171227_014549_14273_sjh9g,
FINISHED,
19 nodes Splits: 175 total,
175 done(
  100.00 %
) 2: 31 [6B ROWS,
18.3GB] [39.7M ROWS / s,
124MB / s]

Needle-in-a-Haystack Lookup on the Hash Key

The largest improvements – 5x, 10x, or more – will be on lookup or filter operations where the partition key columns are tested for equality. Only partitions in the bucket from hashing the partition keys are scanned.

For example, consider:

• table customers is bucketed on customer_id
• table contacts is bucketed on country_code and area_code

These queries will improve:

SELECT... FROM customers WHERE customer_id = 10001;

Here UDP Presto scans only one bucket (the one that 10001 hashes to) if customer_id is the only bucketing key.

SELECT... FROM contacts WHERE country_code='1' and area_code = '650' and phone like'555-____'; 

Here UDP Presto scans only the bucket that matches the hash of country_code 1 + area_code 650.

These queries will not improve:

SELECT... FROM customers WHERE customer_id >= 10001; 

Here UDP will not improve performance, because the predicate doesn't use '='.

SELECT... FROM contacts WHERE area_code = '650' ; 

Here UDP will not improve performance, because the predicate does not include both bucketing keys.

Very Large Join Operations

Very large join operations can sometimes run out of memory. Such joins can benefit from UDP. Distributed and colocated joins will use less memory, CPU, and shuffle less data among Presto workers. This may enable you to finish queries that would otherwise run out of resources. To leverage these benefits, you must:

1. Make sure the two tables to be joined are partitioned on the same keys
2. Use equijoin across all the partitioning keys
3. Set the following options on your join using a magic comment:

-- set session join_distribution_type = 'PARTITIONED'
-- set session colocated_join = 'true'

Improving Performance on Skewed Data

When processing a UDP query, Presto ordinarily creates one split of filtering work per bucket (typically 512 splits, for 512 buckets). But if data is not evenly distributed, filtering on skewed bucket could make performance worse -- one Presto worker node will handle the filtering of that skewed set of partitions, and the whole query lags.

To enable higher scan parallelism you can use:

-- set session distributed_bucket='true|false'

When set to true, multiple splits are used to scan the files in a bucket in parallel, increasing performance. The tradeoff is that colocated join is always disabled when distributed_bucket is true. As a result, some operations such as GROUP BY will require shuffling and more memory during execution.

This query hint is most effective with needle-in-a-haystack queries. Even if these queries perform well with the query hint, test performance with and without the query hint in other use cases on those tables to find the best performance tradeoffs.

Presto Join Performance Improvement with Equi-Joins

Sometimes you can use equi-joins to improve performance of the joins in your Presto queries. The equi-join concatenates tables by comparing join keys using the equal (=) operator.

If this comparison becomes complex, the join processing slows down.

For example, if you want to join two tables with date string, ‘2015-10-01’, but one of the tables only has columns for year, month, and day values, you can write the following query to generate date strings:

SELECT a.date, b.name FROM
left_table a
JOIN right_table b
ON a.date = CAST((b.year * 10000 + b.month * 100 + b.day) as VARCHAR)

This query delays the join processing because the join condition involves several expressions.

To make the example query faster, you can push this condition down into a subquery to prepare a join key beforehand:

SELECT a.date, b.name 
FROM
  left_table a
JOIN (
  SELECT
    CAST((b.year * 10000 + b.month * 100 + b.day) as VARCHAR) date,  
# generate join key
    name
  FROM right_table
) b
ON a.date = b.date  # Simple equi-join

In this example, join keys are a.date and b.date str columns. Comparing VARCHAR strings is much faster than comparing VARCHAR and expression results.

Presto Query Simplification Options

The following are some of the ways that you can simplify your Presto queries.

Use WITH statement

If your query becomes complex or deeply nested, try to extract subqueries using WITH clause. For example, the following query that has a nested subquery:

SELECT a, b, c FROM (
   SELECT a, MAX(b) AS b, MIN(c) AS c FROM tbl GROUP BY a
) tbl_alias

can be rewritten as follows:

WITH tbl_alias AS (SELECT a, MAX(b) AS b, MIN(c) AS c FROM tbl GROUP BY a)
SELECT a, b, c FROM tbl_alias

You can also enumerate multiple sub-queries in WITH clause, by using a comma:

WITH tbl1 AS (SELECT a, MAX(b) AS b, MIN(c) AS c FROM tbl GROUP BY a),
     tbl2 AS (SELECT a, AVG(d) AS d FROM another_tbl GROUP BY a)
SELECT tbl1.*, tbl2.* FROM tbl1 JOIN tbl2 ON tbl1.a = tbl2.a

Use WITH statement in CREATE TABLE statement

If your CREATE TABLE query becomes complex or deeply nested, try to extract subqueries using WITH clause. For example, one sub query can be rewritten as follow:

CREATE TABLE tbl_new AS WITH tbl_alias AS (SELECT a, MAX(b) AS b, MIN(c) AS c FROM tbl1)
SELECT a, b, c FROM tbl_alias

You can also enumerate multiple sub-queries in WITH clause as in the following:

CREATE TABLE tbl_new AS WITH tbl_alias1 AS (SELECT a, MAX(b) AS b, MIN(c) AS c FROM tbl1),
                             tbl_alias2 AS (SELECT a, AVG(d) AS d FROM tbl2)
SELECT tbl_alias1.*, tbl2_alias.* FROM tbl_alias1 JOIN tbl_alias2 ON tbl_alias1.a = tbl_alias2.a

Specify GROUP BY targets with numbers

Group by clause requires having the same expression in the SELECT statement:

SELECT TD_TIME_FORMAT(time, 'yyyy-MM-dd HH', 'PDT') hour, count(*) cnt
FROM my_table
GROUP BY TD_TIME_FORMAT(time, 'yyyy-MM-dd HH', 'PDT')  # <-- redundant expression

You can simplify this query by using GROUP BY 1, 2, …:

SELECT TD_TIME_FORMAT(time, 'yyyy-MM-dd HH', 'PDT') hour, count(*) cnt
FROM my_table
GROUP BY 1

These numbers correspond to the column indexes (1-origin) of the SELECT statement.