Fault-tolerant execution#
By default, if a Trino node lacks the resources to execute a task or otherwise fails during query execution, the query fails and must be run again manually. The longer the runtime of a query, the more likely it is to be susceptible to such failures.
Fault-tolerant execution is a mechanism in Trino that enables a cluster to mitigate query failures by retrying queries or their component tasks in the event of failure. With fault-tolerant execution enabled, intermediate exchange data is spooled and can be re-used by another worker in the event of a worker outage or other fault during query execution.
Note
Fault tolerance does not apply to broken queries or other user error. For example, Trino does not spend resources retrying a query that fails because its SQL cannot be parsed.
Configuration#
Fault-tolerant execution is disabled by default. To enable the feature, set the
retry-policy configuration property to either QUERY or TASK
depending on the desired retry policy.
retry-policy=QUERY
The following configuration properties control the behavior of fault-tolerant execution on a Trino cluster:
Property name |
Description |
Default value |
|---|---|---|
|
Configures what is retried in the event of failure, either
|
|
|
Size of the coordinator’s in-memory buffer used by fault-tolerant execution to store output of query stages. If this buffer is filled during query execution, the query fails unless an exchange manager is configured. |
|
Retry policy#
The retry-policy configuration property designates whether Trino retries
entire queries or a query’s individual tasks in the event of failure.
QUERY#
A QUERY retry policy instructs Trino to automatically retry a query in the
event of an error occuring on a worker node. A QUERY retry policy is
recommended when the majority of the Trino cluster’s workload consists of many
small queries, or if an exchange manager is not
configured.
By default Trino does not implement fault tolerance for queries whose result set
exceeds 32MB in size, such as SELECT statements that return a very
large data set to the user. This limit can be increased by modifying the
exchange.deduplication-buffer-size configuration property to be greater than
the default value of 32MB, but this results in higher memory usage on the
coordinator.
To enable fault-tolerant execution on queries with a larger result set, it is strongly recommended to configure an exchange manager that utilizes external storage for spooled data and therefore allows for storage of spilled data beyond the in-memory buffer size.
TASK#
A TASK retry policy instructs Trino to retry individual query
tasks in the event of failure. This policy is
recommended when executing large batch queries, as the cluster can more
efficiently retry smaller tasks within the query rather than retry the whole
query.
TASK retry policy requires a configured exchange manager to store spooled exchange data used for each task. It is
also strongly recommended to set the query.low-memory-killer.policy
configuration property to total-reservation-on-blocked-nodes, or queries may
need to be manually killed if the cluster runs out of memory.
Advanced configuration#
You can further configure fault-tolerant execution with the following configuration properties. The default values for these properties should work for most deployments, but you can change these values for testing or troubleshooting purposes.
Retry limits#
The following configuration properties control the thresholds at which queries/tasks are no longer retried in the event of repeated failures:
Property name |
Description |
Default value |
Retry policy |
|---|---|---|---|
|
Maximum number of times Trino may attempt to retry a query before declaring the query as failed. |
|
Only |
|
Maximum number retries across all tasks within a given query before declaring the query as failed. |
|
Only |
|
Maximum number of times Trino may attempt to retry a single task before declaring the query as failed. |
|
Only |
|
Minimum time that a failed query or task must wait before it is retried. May be
overridden with the |
|
|
|
Maximum time that a failed query or task must wait before it is retried.
Wait time is increased on each subsequent failure. May be
overridden with the |
|
|
|
Factor by which retry delay is increased on each query or task failure. May be
overridden with the |
|
|
Task sizing#
With a TASK retry policy, it is important to manage the amount of data
processed in each task. If tasks are too small, the management of task
coordination can take more processing time and resources than executing the task
itself. If tasks are too large, then a single task may require more resources
than are available on any one node and therefore prevent the query from
completing.
Trino supports limited automatic task sizing. If issues are occurring
during fault-tolerant task execution, you can configure the following
configuration properties to manually control task sizing. These configuration
properties only apply to a TASK retry policy.
Property name |
Description |
Default value |
|---|---|---|
|
Target size in bytes of all task inputs for a single fault-tolerant task. Applies to tasks that read input from spooled data written by other tasks. May be overridden for the current session with the
|
|
|
Target number of standard splits processed by a single task that reads data from source tables. Value is interpreted with split weight taken into account. If the weight of splits produced by a catalog denotes that they are lighter or heavier than “standard” split, then the number of splits processed by single task is adjusted accordingly. May be overridden for the current session with the
|
|
|
Minimum number of splits processed by a single task. This value is not split weight-adjusted and serves as protection against situations where catalogs report an incorrect split weight. May be overridden for the current session with the
|
|
|
Maximum number of splits processed by a single task. This value is not split weight-adjusted and serves as protection against situations where catalogs report an incorrect split weight. May be overridden for the current session with the
|
|
Node allocation#
With a TASK retry policy, nodes are allocated to tasks based on available
memory and estimated memory usage. If task failure occurs due to exceeding
available memory on a node, the task is restarted with a request to allocate the
full node for its execution.
The initial task memory-requirements estimation is static and configured with
the fault-tolerant-task-memory configuration property. This property only
applies to a TASK retry policy.
Property name |
Description |
Default value |
|---|---|---|
|
Initial task memory estimation used for bin-packing when allocating nodes
for tasks. May be overridden for the current session with the
|
|
Other tuning#
The following additional configuration property can be used to manage fault-tolerant execution:
Property name |
Description |
Default value |
Retry policy |
|---|---|---|---|
|
Maximum amount of memory to be used to store task descriptors for fault tolerant queries on coordinator. Extra memory is needed to be able to reschedule tasks in case of a failure. |
(JVM heap size * 0.15) |
Only |
|
Allow for up to configured number of tasks to wait for node allocation per stage, before pausing scheduling for other tasks from this stage. |
5 |
Only |
Exchange manager#
Exchange spooling is responsible for storing and managing spooled data for fault-tolerant execution. You can configure a filesystem-based exchange manager that stores spooled data in a specified location, either an S3-compatible storage system or a local filesystem.
To configure an exchange manager, create a new
etc/exchange-manager.properties configuration file on the coordinator and
all worker nodes. In this file, set the exchange-manager.name configuration
propertry to filesystem, and additional configuration properties as needed
for your storage solution.
Property name |
Description |
Default value |
|---|---|---|
|
The base directory URI location that the exchange manager uses to store spooling data. Only supports S3 and local filesystems. |
|
|
Enable encrypting of spooling data. |
|
|
The minimum buffer pool size for an exchange sink. The larger the buffer pool size, the larger the write parallelism and memory usage. |
|
|
The number of buffers per partition in the buffer pool. The larger the buffer pool size, the larger the write parallelism and memory usage. |
|
|
Max size of files written by exchange sinks. |
|
|
The number of concurrent readers to read from spooling storage. The larger the number of concurrent readers, the larger the read parallelism and memory usage. |
|
|
AWS access key to use. Required for a connection to AWS S3, can be ignored for other S3 storage systems. |
|
|
AWS secret key to use. Required for a connection to AWS S3, can be ignored for other S3 storage systems. |
|
|
Region of the S3 bucket. |
|
|
S3 storage endpoint server if using an S3-compatible storage system that is not AWS. If using AWS S3, can be ignored. |
|
|
Maximum number of times the exchange manager’s S3 client should retry a request. |
|
|
Part size for S3 multi-part upload. |
|
The following example exchange-manager.properties configuration specifies an
AWS S3 bucket as the spooling storage destination. Note that the destination
does not have to be in AWS, but can be any S3-compatible storage system.
exchange-manager.name=filesystem
exchange.base-directories=s3n://trino-exchange-manager
exchange.encryption-enabled=true
exchange.s3.region=us-west-1
exchange.s3.aws-access-key=example-access-key
exchange.s3.aws-secret-key=example-secret-key
The following example exchange-manager.properties configuration specifies a
local directory, /tmp/trino-exchange-manager, as the spooling storage
destination.
Note
It is only recommended to use a local filesystem for exchange in standalone, non-production clusters. A local directory can only be used for exchange in a distributed cluster if the exchange directory is shared and accessible from all worker nodes.
exchange-manager.name=filesystem
exchange.base-directories=/tmp/trino-exchange-manager