Query Execution Plans


This page provides concepts about query execution plans and how they are used by Cloud Spanner to perform queries in a distributed environment.

Cloud Spanner uses declarative SQL statements to query its databases. SQL statements define what the user wants without specifying how to obtain the results. A query execution plan is the set of steps for how the results are obtained. For a given SQL statement, there may be multiple ways to obtain the results. The Cloud Spanner query compiler evaluates the different ways to produce a query execution plan that is considered the most efficient. Cloud Spanner then uses the execution plan to retrieve the results.

Conceptually, an execution plan is a tree of relational operators. Each operator reads rows from its input(s) and produces output rows. The result of the operator at the root of the execution is returned as the result of the SQL query.

As an example, this query:

SELECT s.SongName FROM Songs AS s;

results in a query execution plan that can be visualized as:

Example query execution plan

The queries and execution plans in this topic are based on the following database schema:

  SingerId   INT64 NOT NULL,
  FirstName  STRING(1024),
  LastName   STRING(1024),
  SingerInfo BYTES(MAX),
  BirthDate  DATE,
) PRIMARY KEY(SingerId);

CREATE INDEX SingersByFirstLastName ON Singers(FirstName, LastName);

  SingerId        INT64 NOT NULL,
  AlbumId         INT64 NOT NULL,
  AlbumTitle      STRING(MAX),
  MarketingBudget INT64,
) PRIMARY KEY(SingerId, AlbumId),

CREATE INDEX AlbumsByAlbumTitle ON Albums(AlbumTitle);

CREATE INDEX AlbumsByAlbumTitle2 ON Albums(AlbumTitle) STORING (MarketingBudget);

  SingerId  INT64 NOT NULL,
  AlbumId   INT64 NOT NULL,
  TrackId   INT64 NOT NULL,
  SongName  STRING(MAX),
  Duration  INT64,
  SongGenre STRING(25),
) PRIMARY KEY(SingerId, AlbumId, TrackId),

CREATE INDEX SongsBySingerAlbumSongNameDesc ON Songs(SingerId, AlbumId, SongName DESC), INTERLEAVE IN Albums;

CREATE INDEX SongsBySongName ON Songs(SongName);

  VenueId      INT64 NOT NULL,
  SingerId     INT64 NOT NULL,
  ConcertDate  DATE NOT NULL,
  BeginTime    TIMESTAMP,
  EndTime      TIMESTAMP,
  TicketPrices ARRAY<INT64>,
) PRIMARY KEY(VenueId, SingerId, ConcertDate);

You can run queries and retrieve execution plans even if the tables have no data.

Obtaining efficient execution plans is challenging because Cloud Spanner divides data into splits. Splits can move independently from each other and get assigned to different servers, which could be in different physical locations. To evaluate execution plans over the distributed data, Cloud Spanner uses execution based on:

  • local execution of subplans in servers that contain the data
  • orchestration and aggregation of multiple remote executions with aggressive distribution pruning

Cloud Spanner uses the primitive operator distributed union, along with its variants distributed cross apply and distributed outer apply, to enable this model.

Life of a query

A SQL query in Cloud Spanner is first compiled into an execution plan, then it is sent to an initial root server for execution. The root server is chosen so as to minimize the number of hops to reach the data being queried. The root server then:

  • initiates remote execution of subplans (if necessary)
  • waits for results from the remote executions
  • handles any remaining local execution steps such as aggregating results
  • returns results for the query

Remote servers that receive a subplan act as a "root" server for their subplan, following the same model as the topmost root server. The result is a tree of remote executions. Conceptually, query execution flows from top to bottom, and query results are returned from bottom to top.The following diagram shows this pattern:

Conceptual query plan

The following examples illustrate this pattern in more detail.

Aggregate queries

An aggregate query implements GROUP BY queries.

For example, using this query:

SELECT s.SingerId, COUNT(*) AS SongCount
FROM Songs AS s
WHERE s.SingerId < 100
GROUP BY s.SingerId;

These are the results:

| SingerId | SongCount |
|        3 |         1 |
|        2 |         8 |

Conceptually, this is the execution plan:

Aggregate query execution plan

Cloud Spanner sends the execution plan to a root server that coordinates the query execution and performs the remote distribution of subplans.

This execution plan starts with a distributed union, which distributes subplans to remote servers whose splits satisfy SingerId < 100. Local distributed unions, shown later in the plan, represent execution on the remote servers. Each local distributed union evaluates a subquery independently on splits of the Songs table, subject to the filter SingerId < 100. The local distributed unions return results to an aggregate operator. The aggregate operator performs the COUNT aggregation by SingerId and returns results to a serialize result operator. The serialize result operator serializes the results into rows that contain the song count by SingerId. The distributed union then unions all results together and returns the query results.

You can learn more about aggregates at aggregate operator.

Co-located join queries

Interleaved tables are physically stored with their rows of related tables co-located. A co-located join is a join between interleaved tables. Co-located joins can offer performance benefits over joins that require indexes or back joins.

For example, using this query:

SELECT al.AlbumTitle, so.SongName
FROM Albums AS al, Songs AS so
WHERE al.SingerId = so.SingerId AND al.AlbumId = so.AlbumId;

(This query assumes that Songs is interleaved in Albums.)

These are the results:

| AlbumTitle            | SongName                 |
| Nothing To Do With Me | Not About The Guitar     |
| Green                 | The Second Time          |
| Green                 | Starting Again           |
| Green                 | Nothing Is The Same      |
| Green                 | Let Us Get Back Together |
| Green                 | I Knew You Were Magic    |
| Green                 | Blue                     |
| Green                 | 42                       |
| Terrified             | Fight Story              |

This is the execution plan:

Co-located join query execution plan

This execution plan starts with a distributed union, which distributes subplans to remote servers that have splits of the table Albums. Because Songs is an interleaved table of Albums, each remote server is able to execute the entire subplan on each remote server without requiring a join to a different server.

The subplans contain a cross apply. Each cross apply performs a table scan on table Albums to retrieve SingerId, AlbumId, and AlbumTitle. The cross apply then maps output from the table scan to output from an index scan on index SongsBySingerAlbumSongNameDesc, subject to a filter of the SingerId in the index matching the SingerId from the table scan output. Each cross apply sends its results to a serialize result operator which serializes the AlbumTitle and SongName data and returns results to the local distributed unions. The distributed union aggregates results from the local distributed unions and returns them as the query result.

Index and back join queries

The example above used a join on two tables, one interleaved in the other. Execution plans are more complex and less efficient when two tables, or a table and an index, are not interleaved.

Consider an index created with the following command:

CREATE INDEX SongsBySongName ON Songs(SongName)

Use this index in this query:

SELECT s.SongName, s.Duration
FROM Songs@{force_index=SongsBySongName} AS s

These are the results:

| SongName | Duration |
| Blue     |      238 |

This is the execution plan:

Back join query execution plan

The resulting execution plan is complicated because the index SongsBySongName does not contain column Duration. To obtain the Duration value, Cloud Spanner needs to back join the indexed results to the table Songs. This is a join but it is not co-located because the Songs table and the global index SongsBySongName are not interleaved. The resulting execution plan is more complex than the co-located join example because Cloud Spanner performs optimizations to speed up the execution if data isn't co-located.

The top operator is a distributed cross apply. This input side of this operator are batches of rows from the index SongsBySongName that satisfy the predicate STARTS_WITH(s.SongName, "B"). The distributed cross apply then maps these batches to remote servers whose splits contain the Duration data. The remote servers use a table scan to retrieve the Duration column. The table scan uses the filter Condition:($Songs_key_TrackId' = $batched_Songs_key_TrackId), which joins TrackId from the Songs table to TrackId of the rows that were batched from the index SongsBySongName.

The results are aggregated into the final query answer. In turn, the input side of the distributed cross apply contains a distributed union/local distributed union pair to evaluate rows from the index that satisfy the STARTS_WITH predicate.

Consider a slightly different query that doesn't select the s.Duration column:

SELECT s.SongName
FROM Songs@{force_index=SongsBySongName} AS s

This query is able to fully leverage the index as shown in this execution plan:

Simpler query execution plan

The execution plan doesn't require a back join because all the columns requested by the query are present in the index.

What's next

Monitor your resources on the go

Get the Google Cloud Console app to help you manage your projects.

Send feedback about...

Cloud Spanner Documentation