THE SQL Server Blog Spot on the Web

Welcome to SQLblog.com - The SQL Server blog spot on the web Sign in | |
in Search

Paul White: Page Free Space

A technical SQL Server blog from New Zealand.

Improving Partitioned Table Join Performance

The query optimizer does not always choose an optimal strategy when joining partitioned tables. This post looks at an example, showing how a manual rewrite of the query can almost double performance, while reducing the memory grant to almost nothing.

Test Data

The two tables in this example use a common partitioning partition scheme. The partition function uses 41 equal-size partitions:

CREATE PARTITION FUNCTION PFT (integer)
AS RANGE RIGHT
FOR VALUES 
    (
        125000, 250000, 375000, 500000,
        625000, 750000, 875000, 1000000,
        1125000, 1250000, 1375000, 1500000,
        1625000, 1750000, 1875000, 2000000,
        2125000, 2250000, 2375000, 2500000,
        2625000, 2750000, 2875000, 3000000,
        3125000, 3250000, 3375000, 3500000,
        3625000, 3750000, 3875000, 4000000,
        4125000, 4250000, 4375000, 4500000,
        4625000, 4750000, 4875000, 5000000
    );
GO
CREATE PARTITION SCHEME PST
AS PARTITION PFT
ALL TO ([PRIMARY]);

There two tables are:

CREATE TABLE dbo.T1
(
    TID integer NOT NULL IDENTITY(0,1),
    Column1 integer NOT NULL,
    Padding binary(100) NOT NULL DEFAULT 0x,
 
    CONSTRAINT PK_T1
    PRIMARY KEY CLUSTERED (TID)
    ON PST (TID)
);
 
CREATE TABLE dbo.T2
(
    TID integer NOT NULL,
    Column1 integer NOT NULL,
    Padding binary(100) NOT NULL DEFAULT 0x,
 
    CONSTRAINT PK_T2
    PRIMARY KEY CLUSTERED (TID, Column1)
    ON PST (TID)
);

The next script loads 5 million rows into T1 with a pseudo-random value between 1 and 5 for Column1. The table is partitioned on the IDENTITY column TID:

INSERT dbo.T1 WITH (TABLOCKX)
    (Column1)
SELECT
    (ABS(CHECKSUM(NEWID())) % 5) + 1
FROM dbo.Numbers AS N
WHERE n BETWEEN 1 AND 5000000;

In case you don’t already have an auxiliary table of numbers lying around, here’s a script to create one with 10 million rows:

CREATE TABLE dbo.Numbers (n bigint PRIMARY KEY);
 
WITH
  L0   AS(SELECT 1 AS c UNION ALL SELECT 1),
  L1   AS(SELECT 1 AS c FROM L0 AS A CROSS JOIN L0 AS B),
  L2   AS(SELECT 1 AS c FROM L1 AS A CROSS JOIN L1 AS B),
  L3   AS(SELECT 1 AS c FROM L2 AS A CROSS JOIN L2 AS B),
  L4   AS(SELECT 1 AS c FROM L3 AS A CROSS JOIN L3 AS B),
  L5   AS(SELECT 1 AS c FROM L4 AS A CROSS JOIN L4 AS B),
  Nums AS(SELECT ROW_NUMBER() OVER (ORDER BY (SELECT NULL)) AS n FROM L5)
INSERT dbo.Numbers WITH (TABLOCKX)
SELECT TOP (10000000) n 
FROM Nums 
ORDER BY n
OPTION (MAXDOP 1);

Table T1 contains data like this:

T1 sample

Next we load data into table T2. The relationship between the two tables is that table 2 contains ‘n’ rows for each row in table 1, where ‘n’ is determined by the value in Column1 of table T1. There is nothing particularly special about the data or distribution, by the way.

INSERT dbo.T2 WITH (TABLOCKX)
    (TID, Column1)
SELECT
    T.TID,
    N.n
FROM dbo.T1 AS T
JOIN dbo.Numbers AS N
    ON N.n >= 1
    AND N.n <= T.Column1;

Table T2 ends up containing about 15 million rows:

T2 sample

The primary key for table T2 is a combination of TID and Column1. The data is partitioned according to the value in column TID alone.

Partition Distribution

The following query shows the number of rows in each partition of table T1:

SELECT
    PartitionID = CA1.P,
    NumRows = COUNT_BIG(*)
FROM dbo.T1 AS T
CROSS APPLY (VALUES ($PARTITION.PFT(TID))) AS CA1 (P)
GROUP BY CA1.P
ORDER BY CA1.P;

Partition Ditribution T1

There are 40 partitions containing 125,000 rows (40 * 125k = 5m rows). The rightmost partition remains empty. The next query shows the distribution for table 2:

SELECT
    PartitionID = CA1.P,
    NumRows = COUNT_BIG(*)
FROM dbo.T2 AS T
CROSS APPLY (VALUES ($PARTITION.PFT(TID))) AS CA1 (P)
GROUP BY CA1.P
ORDER BY CA1.P;

There are roughly 375,000 rows in each partition (the rightmost partition is also empty):

Partition Ditribution T2

Ok, that’s the test data done.

Test Query and Execution Plan

The task is to count the rows resulting from joining tables 1 and 2 on the TID column:

SET STATISTICS IO ON;
DECLARE @s datetime2 = SYSUTCDATETIME();
 
SELECT COUNT_BIG(*)
FROM dbo.T1 AS T1
JOIN dbo.T2 AS T2
    ON T2.TID = T1.TID;
 
SELECT DATEDIFF(Millisecond, @s, SYSUTCDATETIME());
SET STATISTICS IO OFF;

The optimizer chooses a plan using parallel hash join, and partial aggregation:

Parallel Hash Join Plan

The Plan Explorer plan tree view shows accurate cardinality estimates and an even distribution of rows across threads (click to enlarge the image):

Plan Explorer Plan Tree View 1

With a warm data cache, the STATISTICS IO output shows that no physical I/O was needed, and all 41 partitions were touched:

STATISTICS IO

Running the query without actual execution plan or STATISTICS IO information for maximum performance, the query returns in around 2600ms.

Execution Plan Analysis

The first step toward improving on the execution plan produced by the query optimizer is to understand how it works, at least in outline.

The two parallel Clustered Index Scans use multiple threads to read rows from tables T1 and T2. Parallel scan uses a demand-based scheme where threads are given page(s) to scan from the table as needed. This arrangement has certain important advantages, but does result in an unpredictable distribution of rows amongst threads. The point is that multiple threads cooperate to scan the whole table, but it is impossible to predict which rows end up on which threads.

For correct results from the parallel hash join, the execution plan has to ensure that rows from T1 and T2 that might join are processed on the same thread. For example, if a row from T1 with join key value ‘1234’ is placed in thread 5’s hash table, the execution plan must guarantee that any rows from T2 that also have join key value ‘1234’ probe thread 5’s hash table for matches.

The way this guarantee is enforced in this parallel hash join plan is by repartitioning rows to threads after each parallel scan. The two repartitioning exchanges route rows to threads using a hash function over the hash join keys. The two repartitioning exchanges use the same hash function so rows from T1 and T2 with the same join key must end up on the same hash join thread.

Expensive Exchanges

This business of repartitioning rows between threads can be very expensive, especially if a large number of rows is involved. The execution plan selected by the optimizer moves 5 million rows through one repartitioning exchange and around 15 million across the other.

As a first step toward removing these exchanges, consider the execution plan selected by the optimizer if we join just one partition from each table, disallowing parallelism:

SELECT COUNT_BIG(*)
FROM dbo.T1 AS T1
JOIN dbo.T2 AS T2
    ON T2.TID = T1.TID
WHERE
    $PARTITION.PFT(T1.TID) = 1
    AND $PARTITION.PFT(T2.TID) = 1
OPTION (MAXDOP 1);

Single Partition Plan

The optimizer has chosen a (one-to-many) merge join instead of a hash join. The single-partition query completes in around 100ms. If everything scaled linearly, we would expect that extending this strategy to all 40 populated partitions would result in an execution time around 4000ms. Using parallelism could reduce that further, perhaps to be competitive with the parallel hash join chosen by the optimizer.

This raises a question. If the most efficient way to join one partition from each of the tables is to use a merge join, why does the optimizer not choose a merge join for the full query?

Forcing a Merge Join

Let’s force the optimizer to use a merge join on the test query using a hint:

SELECT COUNT_BIG(*)
FROM dbo.T1 AS T1
JOIN dbo.T2 AS T2
    ON T2.TID = T1.TID
OPTION (MERGE JOIN);

This is the execution plan selected by the optimizer:

Optimizer Merge Join Plan

This plan results in the same number of logical reads reported previously, but instead of 2600ms the query takes 5000ms. The natural explanation for this drop in performance is that the merge join plan is only using a single thread, whereas the parallel hash join plan could use multiple threads.

Parallel Merge Join

We can get a parallel merge join plan using the same query hint as before, and adding trace flag 8649:

SELECT COUNT_BIG(*)
FROM dbo.T1 AS T1
JOIN dbo.T2 AS T2
    ON T2.TID = T1.TID
OPTION (MERGE JOIN, QUERYTRACEON 8649);
The execution plan is:

Optimizer Parallel Merge Join

This looks promising. It uses a similar strategy to distribute work across threads as seen for the parallel hash join. In practice though, performance is disappointing. On a typical run, the parallel merge plan runs for around 8400ms; slower than the single-threaded merge join plan (5000ms) and much worse than the 2600ms for the parallel hash join. We seem to be going backwards!

The logical reads for the parallel merge are still exactly the same as before, with no physical IOs. The cardinality estimates and thread distribution are also still very good (click to enlarge):

Plan Explorer Tree View 2

A big clue to the reason for the poor performance is shown in the wait statistics (captured by Plan Explorer Pro):

Plan Explorer Pro Waits

CXPACKET waits require careful interpretation, and are most often benign, but in this case excessive waiting occurs at the repartitioning exchanges. Unlike the parallel hash join, the repartitioning exchanges in this plan are order-preserving ‘merging’ exchanges (because merge join requires ordered inputs):

Reprartition Streams Operator Properties

Parallelism works best when threads can just grab any available unit of work and get on with processing it. Preserving order introduces inter-thread dependencies that can easily lead to significant waits occurring. In extreme cases, these dependencies can result in an intra-query deadlock, though the details of that will have to wait for another time to explore in detail.

The potential for waits and deadlocks leads the query optimizer to cost parallel merge join relatively highly, especially as the degree of parallelism (DOP) increases. This high costing resulted in the optimizer choosing a serial merge join rather than parallel in this case. The test results certainly confirm its reasoning.

Collocated Joins

The optimizer has another available strategy when joining tables that share a common partition scheme. This strategy is a collocated join, also known as as a per-partition join. It can be applied in both serial and parallel execution plans, though it is limited to 2-way joins in the current optimizer. Whether the optimizer chooses a collocated join or not depends on cost estimation. The primary benefits of a collocated join are that it eliminates an exchange and requires less memory, as we will see next.

Costing and Plan Selection

The query optimizer did consider a collocated join for our original query, but it was rejected on cost grounds. The parallel hash join with repartitioning exchanges appeared to be a cheaper option. There is no query hint to force a collocated join, so we have to mess with the costing framework to produce one for our test query. Pretending that IOs cost 50 times more than usual is enough to convince the optimizer to use collocated join with our test query:

-- Pretend IOs are 50x cost temporarily
DBCC SETIOWEIGHT(50);
 
-- Co-located hash join
SELECT COUNT_BIG(*)
FROM dbo.T1 AS T1
JOIN dbo.T2 AS T2
    ON T2.TID = T1.TID
OPTION (RECOMPILE);
 
-- Reset IO costing
DBCC SETIOWEIGHT(1);

Collocated Join Plan

The estimated execution plan for the collocated join is:

Collocated Hash Join

The Constant Scan contains one row for each partition of the shared partitioning scheme, from 1 to 41. The hash repartitioning exchanges seen previously are replaced by a single Distribute Streams exchange using Demand partitioning. Demand partitioning means that the next partition id is given to the next parallel thread that asks for one.

My test machine has eight logical processors, and all are available for SQL Server to use. As a result, there are eight threads in the single parallel branch in this plan, each processing one partition from each table at a time. Once a thread finishes processing a partition, it grabs a new partition number from the Distribute Streams exchange…and so on until all partitions have been processed.

It is important to understand that the parallel scans in this plan are different from the parallel hash join plan. Although the scans have the same parallelism icon, tables T1 and T2 are not being co-operatively scanned by multiple threads in the same way. Each thread reads a single partition of T1 and performs a hash match join with the same partition from table T2. The properties of the two Clustered Index Scans show a Seek Predicate (unusual for a scan!) limiting the rows to a single partition:

Clustered Index Scan Properties

The crucial point is that the join between T1 and T2 is on TID, and TID is the partitioning column for both tables. A thread that processes partition ‘n’ is guaranteed to see all rows that can possibly join on TID for that partition. In addition, no other thread will see rows from that partition, so this removes the need for repartitioning exchanges.

CPU and Memory Efficiency Improvements

The collocated join has removed two expensive repartitioning exchanges and added a single exchange processing 41 rows (one for each partition id). Remember, the parallel hash join plan exchanges had to process 5 million and 15 million rows. The amount of processor time spent on exchanges will be much lower in the collocated join plan.

In addition, the collocated join plan has a maximum of 8 threads processing single partitions at any one time. The 41 partitions will all be processed eventually, but a new partition is not started until a thread asks for it. Threads can reuse hash table memory for the new partition.

The parallel hash join plan also had 8 hash tables, but with all 5,000,000 build rows loaded at the same time. The collocated plan needs memory for only 8 * 125,000 = 1,000,000 rows at any one time.

Collocated Hash Join Performance

The collated join plan has disappointing performance in this case. The query runs for around 25,300ms despite the same IO statistics as usual. This is much the worst result so far, so what went wrong?

It turns out that cardinality estimation for the single partition scans of table T1 is slightly low. The properties of the Clustered Index Scan of T1 (graphic immediately above) show the estimation was for 121,951 rows. This is a small shortfall compared with the 125,000 rows actually encountered, but it was enough to cause the hash join to spill to physical tempdb:

Collocated Hash Join Performance

A level 1 spill doesn’t sound too bad, until you realize that the spill to tempdb probably occurs for each of the 41 partitions.

As a side note, the cardinality estimation error is a little surprising because the system tables accurately show there are 125,000 rows in every partition of T1. Unfortunately, the optimizer uses regular column and index statistics to derive cardinality estimates here rather than system table information (e.g. sys.partitions).

Collocated Merge Join

We will never know how well the collocated parallel hash join plan might have worked without the cardinality estimation error (and the resulting 41 spills to tempdb) but we do know:

  • Merge join does not require a memory grant; and
  • Merge join was the optimizer’s preferred join option for a single partition join

Putting this all together, what we would really like to see is the same collocated join strategy, but using merge join instead of hash join. Unfortunately, the current query optimizers (SQL Server 2008 and later) cannot produce a collocated merge join; it only knows how to do collocated hash join. SQL Server 2005 used an APPLY pattern to join partitioned tables, so a collocated merge join was possible in that version.

So where does this leave us?

CROSS APPLY sys.partitions

We can try to write our own collocated join query. We can use sys.partitions to find the partition numbers, and CROSS APPLY to get a count per partition, with a final step to sum the partial counts. The following query implements this idea:

SELECT
    row_count = SUM(Subtotals.cnt)
FROM
(
    -- Partition numbers
    SELECT
        p.partition_number
    FROM sys.partitions AS p 
    WHERE 
        p.[object_id] = OBJECT_ID(N'T1', N'U')
        AND p.index_id = 1
) AS P
CROSS APPLY
(
    -- Count per collocated join
    SELECT
        cnt = COUNT_BIG(*)
    FROM dbo.T1 AS T1
    JOIN dbo.T2 AS T2
        ON T2.TID = T1.TID
    WHERE 
        $PARTITION.PFT(T1.TID) = p.partition_number
        AND $PARTITION.PFT(T2.TID) = p.partition_number
) AS SubTotals;

The estimated plan is:

Partition Apply

The cardinality estimates aren’t all that good here, especially the estimate for the scan of the system table underlying the sys.partitions view.

Nevertheless, the plan shape is heading toward where we would like to be. Each partition number from the system table results in a per-partition scan of T1 and T2, a one-to-many Merge Join, and a Stream Aggregate to compute the partial counts. The final Stream Aggregate just sums the partial counts.

Execution time for this query is around 3,500ms, with the same IO statistics as always. This compares favourably with 5,000ms for the serial plan produced by the optimizer with the OPTION (MERGE JOIN) hint. This is another case of the sum of the parts being less than the whole – summing 41 partial counts from 41 single-partition merge joins is faster than a single merge join and count over all partitions.

Even so, this single-threaded collocated merge join is not as quick as the original parallel hash join plan, which executed in 2,600ms.

On the positive side, our collocated merge join uses only one logical processor and requires no memory grant. The parallel hash join plan used 16 threads and reserved 569 MB of memory:

image 

Using a Temporary Table

Our collocated merge join plan should benefit from parallelism. The reason parallelism is not being used is that the query references a system table. We can work around that by writing the partition numbers to a temporary table (or table variable):

SET STATISTICS IO ON;
DECLARE @s datetime2 = SYSUTCDATETIME();
 
CREATE TABLE #P
(
    partition_number integer PRIMARY KEY);
 
INSERT #P (partition_number)
SELECT
    p.partition_number
FROM sys.partitions AS p 
WHERE 
    p.[object_id] = OBJECT_ID(N'T1', N'U')
    AND p.index_id = 1;
 
SELECT
    row_count = SUM(Subtotals.cnt)
FROM #P AS p
CROSS APPLY
(
    SELECT
        cnt = COUNT_BIG(*)
    FROM dbo.T1 AS T1
    JOIN dbo.T2 AS T2
        ON T2.TID = T1.TID
    WHERE 
        $PARTITION.PFT(T1.TID) = p.partition_number
        AND $PARTITION.PFT(T2.TID) = p.partition_number
) AS SubTotals;
 
DROP TABLE #P;
 
SELECT DATEDIFF(Millisecond, @s, SYSUTCDATETIME());
SET STATISTICS IO OFF;

Using the temporary table adds a few logical reads, but the overall execution time is still around 3500ms, indistinguishable from the same query without the temporary table. The problem is that the query optimizer still doesn’t choose a parallel plan for this query, though the removal of the system table reference means that it could if it chose to:

Temporary Table Plan

In fact the optimizer did enter the parallel plan phase of query optimization (running search 1 for a second time):

Optimizer Search Phases

Unfortunately, the parallel plan found seemed to be more expensive than the serial plan. This is a crazy result, caused by the optimizer’s cost model not reducing operator CPU costs on the inner side of a nested loops join. Don’t get me started on that, we’ll be here all night.

In this plan, everything expensive happens on the inner side of a nested loops join. Without a CPU cost reduction to compensate for the added cost of exchange operators, candidate parallel plans always look more expensive to the optimizer than the equivalent serial plan.

Parallel Collocated Merge Join

We can produce the desired parallel plan using trace flag 8649 again:

SELECT
    row_count = SUM(Subtotals.cnt)
FROM #P AS p
CROSS APPLY
(
    SELECT
        cnt = COUNT_BIG(*)
    FROM dbo.T1 AS T1
    JOIN dbo.T2 AS T2
        ON T2.TID = T1.TID
    WHERE 
        $PARTITION.PFT(T1.TID) = p.partition_number
        AND $PARTITION.PFT(T2.TID) = p.partition_number
) AS SubTotals
OPTION (QUERYTRACEON 8649);

The actual execution plan is:

Parallel Collocated Merge Join

One difference between this plan and the collocated hash join plan is that a Repartition Streams exchange operator is used instead of Distribute Streams. The effect is similar, though not quite identical. The Repartition uses round-robin partitioning, meaning the next partition id is pushed to the next thread in sequence. The Distribute Streams exchange seen earlier used Demand partitioning, meaning the next partition id is pulled across the exchange by the next thread that is ready for more work. There are subtle performance implications for each partitioning option, but going into that would again take us too far off the main point of this post.

Performance

The important thing is the performance of this parallel collocated merge join – just 1350ms on a typical run.

The list below shows all the alternatives from this post (all timings include creation, population, and deletion of the temporary table where appropriate) from quickest to slowest:

  • Collocated parallel merge join: 1350ms
  • Parallel hash join: 2600ms
  • Collocated serial merge join: 3500ms
  • Serial merge join: 5000ms
  • Parallel merge join: 8400ms
  • Collated parallel hash join: 25,300ms (hash spill per partition)

The parallel collocated merge join requires no memory grant (aside from a paltry 1.2MB used for exchange buffers). This plan uses 16 threads at DOP 8; but 8 of those are (rather pointlessly) allocated to the parallel scan of the temporary table. These are minor concerns, but it turns out there is a way to address them if it bothers you.

Parallel Collocated Merge Join with Demand Partitioning

This final tweak replaces the temporary table with a hard-coded list of partition ids (dynamic SQL could be used to generate this query from sys.partitions):

SELECT
    row_count = SUM(Subtotals.cnt)
FROM
(
    VALUES 
        (1),(2),(3),(4),(5),(6),(7),(8),(9),(10),
        (11),(12),(13),(14),(15),(16),(17),(18),(19),(20),
        (21),(22),(23),(24),(25),(26),(27),(28),(29),(30),
        (31),(32),(33),(34),(35),(36),(37),(38),(39),(40),(41)
) AS P (partition_number)
CROSS APPLY
(
    SELECT
        cnt = COUNT_BIG(*)
    FROM dbo.T1 AS T1
    JOIN dbo.T2 AS T2
        ON T2.TID = T1.TID
    WHERE 
        $PARTITION.PFT(T1.TID) = p.partition_number
        AND $PARTITION.PFT(T2.TID) = p.partition_number
) AS SubTotals
OPTION (QUERYTRACEON 8649);

The actual execution plan is:

Parallel Collocated Merge Join - Demand

The parallel collocated hash join plan is reproduced below for comparison:

Parallel Collocated Hash Join

The manual rewrite has another advantage that has not been mentioned so far: the partial counts (per partition) can be computed earlier than the partial counts (per thread) in the optimizer’s collocated join plan. The earlier aggregation is performed by the extra Stream Aggregate under the nested loops join.

The performance of the parallel collocated merge join is unchanged at around 1350ms.

Final Words

It is a shame that the current query optimizer does not consider a collocated merge join (Connect item closed as Won’t Fix). The example used in this post showed an improvement in execution time from 2600ms to 1350ms using a modestly-sized data set and limited parallelism. In addition, the memory requirement for the query was almost completely eliminated  – down from 569MB to 1.2MB.

The problem with the parallel hash join selected by the optimizer is that it attempts to process the full data set all at once (albeit using eight threads). It requires a large memory grant to hold all 5 million rows from table T1 across the eight hash tables, and does not take advantage of the divide-and-conquer opportunity offered by the common partitioning.

The great thing about the collocated join strategies is that each parallel thread works on a single partition from both tables, reading rows, performing the join, and computing a per-partition subtotal, before moving on to a new partition.

Note that the general technique is not limited to partitioned tables. Any table that can be naturally segmented using a driver table of values can be written to use a collocated merge join, by substituting a driver value for the partition id.

From a thread’s point of view…

If you have trouble visualizing what is happening from just looking at the parallel collocated merge join execution plan, let’s look at it again, but from the point of view of just one thread operating between the two Parallelism (exchange) operators.

image

Our thread picks up a single partition id from the Distribute Streams exchange, and starts a merge join using ordered rows from partition 1 of table T1 and partition 1 of table T2. By definition, this is all happening on a single thread. As rows join, they are added to a (per-partition) count in the Stream Aggregate immediately above the Merge Join. Eventually, either T1 (partition 1) or T2 (partition 1) runs out of rows and the merge join stops.

The per-partition count from the aggregate passes on through the Nested Loops join to another Stream Aggregate, which is maintaining a per-thread subtotal. Our same thread now picks up a new partition id from the exchange (say it gets id 9 this time). The count in the per-partition aggregate is reset to zero, and the processing of partition 9 of both tables proceeds just as it did for partition 1, and on the same thread.

Each thread picks up a single partition id and processes all the data for that partition, completely independently from other threads working on other partitions. One thread might eventually process partitions (1, 9, 17, 25, 33, 41) while another is concurrently processing partitions (2, 10, 18, 26, 34) and so on for the other six threads at DOP 8.

The point is that all 8 threads can execute independently and concurrently, continuing to process new partitions until the wider job (of which the thread has no knowledge!) is done. This divide-and-conquer technique can be much more efficient than simply splitting the entire workload across eight threads all at once.

PASS_2013_SpeakingButton_180x180

Related Reading

Understanding and Using Parallelism in SQL Server

Parallel Execution Plans Suck

© 2013 Paul White – All Rights Reserved

Twitter: @SQL_Kiwi
Email: SQLkiwi@gmail.com

Published Monday, June 17, 2013 5:22 AM by Paul White

Comment Notification

If you would like to receive an email when updates are made to this post, please register here

Subscribe to this post's comments using RSS

Comments

 

Jason Horner said:

Great post Paul. Wondering if we used a local partition view instead of the built-in table partitioning if we would still be susceptible to the extra parallel scans, and eliminate the need for dynamic sql.

June 16, 2013 12:09 PM
 

Ash said:

Awesome!!!Always tons of good stuff in your posts that will have me bookmarking it for re-reads....

June 16, 2013 12:47 PM
 

Paul White said:

Hi Jason,

Yes it is possible to do something similar with local partitioned views, though care is needed to get the best from it.

Paul

June 17, 2013 6:39 AM
 

cdp said:

Hi Paul,

I was following your example on my own but the data/content for your numbers table seems to be missing. Can you let me know table design and content?

Thanks,

Chris.

June 18, 2013 11:27 AM
 

Paul White said:

CDP,

It's a fairly standard numbers table, I'll update the post to include the code for it.

Paul 

June 18, 2013 12:18 PM
 

Chris Adkin said:

Hi Paul,

Again another excellent post, I would really like to see you deliver some training / a pre-con in Europe. Going off at a slight tangent, and I appreciate you data does not fit this scenario, but is it possible to force a parallel few outer rows join ?.

Chris

June 20, 2013 8:14 AM
 

Paul White said:

Hi Chris,

If the opportunity presents itself one day, I would certainly consider it - thanks for saying that.

As far as forcing the round-robin repartition for the 'few outer rows optimization' is concerned, I'm not aware of anything (like a specific trace flag) to force it, so we're left with tricks like TOP (@n) with OPTIMIZE FOR to make it look as if the number of rows is in the range where this optimization is applied. It's a hack of course, and vulnerable to future changes in optimizer behaviour, but it's the best we have today, as far as I know. Other T-SQL tricks that don't change the semantics but result in an in-range cardinality estimate are possible too, naturally.

Paul

June 20, 2013 9:05 AM
 

Adam Machanic said:

Paul,

Any clues on why the QO would push the partition elimination predicate into a filter AFTER the scan, instead of using it as a seek predicate on the scan? And how to avoid such issues? (Preferably without any query hints -- I'm trying to pack this into a view.)

The result of the QO's decision in this case means that instead of getting great performance I'm getting one scan of the entire table, per partition. Pretty abysmal :-(

Thanks,

Adam

July 3, 2013 10:02 AM
 

Paul White said:

Hi Adam,

I have seen the optimizer choose a separate filter, but not in exactly the circumstances you seem to be describing. Aside from a FORCESEEK hint, it's hard to know what to suggest without seeing the execution plan. Perhaps you are driving the collocated join using partition values instead of ids? Anyway, if you can send me the plan, I'll be happy to take a look.

Cheers,

Paul

July 3, 2013 7:30 PM
 

Tom said:

Hi Paul,

I'm working through your example. I noticed that when I run the test query, it uses a merge join by default rather than the hash match with parallelism repartition streams. Any idea why this would happen? I have double-checked the data is setup the same, similar number of rows in partitions, etc. I have tried this on 2008R2 and 2012 DBs and it's the same behaviour. I've checked settings such as maxdop and it's defaulted to 0.

Cheers,

Tom

July 8, 2013 6:56 AM
 

Paul White said:

Hi Tom,

There is an element of chance with the data distribution, and the choice between parallel hash and serial merge plans in this case is also quite finely balanced (according to the cost model anyway).

To see the parallel plan, you have a couple of choices. 'Force' parallelism using an OPTION (QUERYTRACEON 8649) clause on the query, or pretend you have more logical processing units available using DBCC OPTIMIZER_WHATIF(CPUs, 16). More logical processors often tilts plan choice toward parallelism due to its effect on costing.

Either of those should produce the parallel hash plan. If you choose to use the DBCC command, reset it back to the default setting with DBCC OPTIMIZER_WHATIF(CPUs, 0) after you have finished testing.

Paul

July 8, 2013 7:27 AM
 

JRStern said:

You're a wizard, Harry ... er, Paul!

I'm not sure what's scarier here, the undocumented dbcc's, or optimizing a partitioned query by referencing partition ids explicitly.  Wish I'd ever noticed your description of the SETIOWEIGHT before, could have used that recently, but I expect I'll find more uses for it soon enough.

Of course what's really scary is that SQL Server needs so much coercion to get any of this right, one might have thought this was all going to be baked in when the partitioned tables were introduced ... but one would then have been wrong.

Well I'm off tomorrow to talk to a place using partitioned tables bigtime on about two terabytes (uncompressed) with a heavy daily (monthly?) ETL updated.  Apparently they don't trust TSQL to get it at all right, so they (re)process the partitions individually then roll them back in.  Maybe some of these tricks would help - but their current strategy is apparently to let SSIS do the hard work.  Just making conversation, what do you think of that, using SSIS merge and SCD tasks rather than TSQL, especially in and around partitioned tables?

Thanks.

July 26, 2013 2:11 AM
 

Paul White said:

Hi JRStern,

I agree there is plenty more that could be done regarding query optimization, parallelism, and query plans for partitioned tables in general.

The idea of hiding the complexity of partitioned tables using the query processor is an interesting design choice; it definitely needs more engineering investment to achieve its full potential.

The balance of history is against us here though; features rarely get the sort of incremental improvement in later versions that we might expect. Often, backward compatibility and concern over sudden plan changes are cited reasons. Meh.

It's hard to generalize about a system I haven't seen, but I have come across good implementations using T-SQL and using SSIS (and bad ones for both, naturally). The default SCD component is notoriously inefficient, however. I would be surprised if there weren't scope for you to improve their current process. Processing single partitions and using SWITCH could work very well, the details matter though.

July 26, 2013 2:56 AM
 

marek said:

Hello,

really interesting post

what do you mean by '...the general technique is not limited to partitioned tables...'

I can't think about any other example

August 9, 2013 3:10 AM
 

Paul White said:

Hi Marek,

The same technique can be used where a 'section' of each table can be joined one at a time. The basic idea is to use an APPLY driven by a table that contains section identifiers. So, instead of joining a partition at a time, we join a section at a time.

For example, the driving table might be a list of customer IDs, which is then applied to a join of two or more tables that contain the customer ID as part of the join criteria.

Paul

August 12, 2013 10:16 PM
 

marek said:

Thank you Paul,

I thought that. Anyway I've seen Adam Machanic Manhandling Paralellism

lecture from PASS Summit and I tried to use it, all techniques that were mentioned and the result is: simple hash match join for me still works better and was almost always choosed by QO, I dont know why.

Only way I was able to force QO to use nested loop was cross apply with TOP, distinct and 0%[primary_key]+[primary_key] on driving table

and top on inner side. It was slower. Yeah and I also thought till I saw it, that cross apply always forces nested loop

August 14, 2013 4:00 AM
 

Paul White said:

Yes, the costing model does discriminate against parallel nested loops join. It can be quite an exercise to write the query in a way that produces the desired plan with *all* the necessary features.

Cross apply is a logical operation that can be transformed to a join (and then implemented by hash or sort-merge) by the optimizer. It so happens that outer apply is much less easily transformed, but it is possible, and that state of affairs could change over time.

If you have a particular query plan you are trying to optimize for parallel execution, I would suggest uploading it for analysis using the SQL Sentry Plan Explorer tool. Post-execution, non-anonymized plans are much easier to analyse.

Paul

August 14, 2013 5:54 AM
 

marek said:

Well, finally I got it (what can you do without proper indexes).

But funny thing is (and again I don't know why) :

thread 0: 0 rows (that's ok)

threads 1-31 nice distribution around 70k  (+/- 1k)

thread 32: 250k

No matter what I do, one thread is still high, on inner side.

Outer side is nicely distributed.

I tried update statistics, changed maxdop, no help.

August 18, 2013 11:42 AM
 

marek said:

waaauu and it propably has a lot to do with uniqueness on inner side join column

August 18, 2013 1:20 PM
 

Paul White said:

Marek,

It might be unavoidable depending on the way outer side rows are being distributed. It sounds like there's a round-robin exchange above the join, and one (or more) correlated values results in more matches on the inner side than for other rows. Impossible to be certain without seeing the plan, but the distribution you have seems 'good enough'.

August 18, 2013 8:21 PM

Leave a Comment

(required) 
(required) 
Submit
Powered by Community Server (Commercial Edition), by Telligent Systems
  Privacy Statement