SQL Server does not optimize parallel merge join on two equivalently partitioned tables

Question

Apologies in advance for the very detailed question. I have included queries to generate a full data set for reproducing the problem, and I am running SQL Server 2012 on a 32-core machine. However, I do not think this is specific to SQL Server 2012, and I have forced a MAXDOP of 10 for this particular example.

I have two tables that are partitioned using the same partition scheme. When joining them together on the column used for partitioning, I noticed that SQL Server is not able to optimize a parallel merge join as much as one might expect and thus chooses to use a HASH JOIN instead. In this particular case, I am able to manually simulate a much more optimal parallel MERGE JOIN by splitting the query into 10 disjoint ranges based on the partition function and running each of those queries simultaneously in SSMS. Using WAITFOR to run them all at precisely the same time, the result is that all of the queries complete in ~40% of the total time used by the original parallel HASH JOIN.

Is there any way to get SQL Server to make this optimization on its own in the case of equivalently partitioned tables? I understand that SQL Server may generally incur a lot of overhead in order to make a MERGE JOIN parallel, but it seems like there is a very natural sharding method with minimal overhead in this case. Perhaps it is just a specialized case that the optimizer isn't yet clever enough to recognize?

Here is the SQL to set up a simplified data set in order to reproduce this problem:

/* Create the first test data table */
CREATE TABLE test_transaction_properties 
    ( transactionID INT NOT NULL IDENTITY(1,1)
    , prop1 INT NULL
    , prop2 FLOAT NULL
    )

/* Populate table with pseudo-random data (the specific data doesn't matter too much for this example) */
;WITH E1(N) AS (
    SELECT 1 UNION ALL SELECT 1 UNION ALL SELECT 1 UNION ALL SELECT 1 UNION ALL SELECT 1 
    UNION ALL SELECT 1 UNION ALL SELECT 1 UNION ALL SELECT 1 UNION ALL SELECT 1 UNION ALL SELECT 1
)
, E2(N) AS (SELECT 1 FROM E1 a CROSS JOIN E1 b)
, E4(N) AS (SELECT 1 FROM E2 a CROSS JOIN E2 b)
, E8(N) AS (SELECT 1 FROM E4 a CROSS JOIN E4 b)
INSERT INTO test_transaction_properties WITH (TABLOCK) (prop1, prop2)
SELECT TOP 10000000 (ABS(CAST(CAST(NEWID() AS VARBINARY) AS INT)) % 5) + 1 AS prop1
                , ABS(CAST(CAST(NEWID() AS VARBINARY) AS INT)) * rand() AS prop2
FROM E8

/* Create the second test data table */
CREATE TABLE test_transaction_item_detail
    ( transactionID INT NOT NULL
    , productID INT NOT NULL
    , sales FLOAT NULL
    , units INT NULL
    )

 /* Populate the second table such that each transaction has one or more items
     (again, the specific data doesn't matter too much for this example) */
INSERT INTO test_transaction_item_detail WITH (TABLOCK) (transactionID, productID, sales, units)
SELECT t.transactionID, p.productID, 100 AS sales, 1 AS units
FROM test_transaction_properties t
JOIN (
    SELECT 1 as productRank, 1 as productId
    UNION ALL SELECT 2 as productRank, 12 as productId
    UNION ALL SELECT 3 as productRank, 123 as productId
    UNION ALL SELECT 4 as productRank, 1234 as productId
    UNION ALL SELECT 5 as productRank, 12345 as productId
) p
    ON p.productRank <= t.prop1

/* Divides the transactions evenly into 10 partitions */
CREATE PARTITION FUNCTION [pf_test_transactionId] (INT)
AS RANGE RIGHT
FOR VALUES
(1,1000001,2000001,3000001,4000001,5000001,6000001,7000001,8000001,9000001)

CREATE PARTITION SCHEME [ps_test_transactionId]
AS PARTITION [pf_test_transactionId]
ALL TO ( [PRIMARY] )

/* Apply the same partition scheme to both test data tables */
ALTER TABLE test_transaction_properties
ADD CONSTRAINT PK_test_transaction_properties
PRIMARY KEY (transactionID)
ON ps_test_transactionId (transactionID)

ALTER TABLE test_transaction_item_detail
ADD CONSTRAINT PK_test_transaction_item_detail
PRIMARY KEY (transactionID, productID)
ON ps_test_transactionId (transactionID)

Now we are finally ready to reproduce the sub-optimal query!

/* This query produces a HASH JOIN using 20 threads without the MAXDOP hint,
    and the same behavior holds in that case.
    For simplicity here, I have limited it to 10 threads. */
SELECT COUNT(*)
FROM test_transaction_item_detail i
JOIN test_transaction_properties t
    ON t.transactionID = i.transactionID
OPTION (MAXDOP 10)

However, using a single thread to process each partition (example for first partition below) would lead to a much more efficient plan. I tested this by running a query like the one below for each of the 10 partitions at precisely the same moment, and all 10 finished in just over 1 second:

SELECT COUNT(*)
FROM test_transaction_item_detail i
INNER MERGE JOIN test_transaction_properties t
    ON t.transactionID = i.transactionID
WHERE t.transactionID BETWEEN 1 AND 1000000
OPTION (MAXDOP 1)

Answer 1

You're right that the SQL Server optimizer prefers not to generate parallel MERGE join plans (it costs this alternative quite high). Parallel MERGE always requires repartitioning exchanges on both join inputs, and more importantly, it requires that row order be preserved across those exchanges.

Parallelism is most efficient when each thread can run independently; order preservation often leads to frequent synchronization waits, and may ultimately cause exchanges to spill to tempdb to resolve an intra-query deadlock condition.

These problems can be circumvented by running multiple instances of the whole query on one thread each, with each thread processing an exclusive range of data. This is not a strategy that the optimizer considers natively, however. As it is, the original SQL Server model for parallelism breaks the query at exchanges, and runs the plan segments formed by those splits on multiple threads.

There are ways to achieve running whole query plans on multiple threads over exclusive data set ranges, but they require trickery that not everyone will be happy with (and will not be supported by Microsoft or guaranteed to work in the future). One such approach is to iterate over the partitions of a partitioned table and give each thread the task of producing a subtotal. The result is the SUM of the row counts returned by each independent thread:

Obtaining partition numbers is easy enough from metadata:

DECLARE @P AS TABLE
(
    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'test_transaction_properties', N'U')
    AND p.index_id = 1;

We then use these numbers to drive a correlated join (APPLY), and the $PARTITION function to limit each thread to the current partition number:

SELECT
    row_count = SUM(Subtotals.cnt)
FROM @P AS p
CROSS APPLY
(
    SELECT
        cnt = COUNT_BIG(*)
    FROM dbo.test_transaction_item_detail AS i
    JOIN dbo.test_transaction_properties AS t ON
        t.transactionID = i.transactionID
    WHERE 
        $PARTITION.pf_test_transactionId(t.transactionID) = p.partition_number
        AND $PARTITION.pf_test_transactionId(i.transactionID) = p.partition_number
) AS SubTotals;

The query plan shows a MERGE join being performed for each row in table @P. The clustered index scan properties confirm that only a single partition is processed on each iteration:

Unfortunately, this only results in sequential serial processing of partitions. On the data set you provided, my 4-core (hyperthreaded to 8) laptop returns the correct result in 7 seconds with all data in memory.

To get the MERGE sub-plans to run concurrently, we need a parallel plan where partition ids are distributed over the available threads (MAXDOP) and each MERGE sub-plan runs on a single thread using the data in one partition. Unfortunately, the optimizer frequently decides against parallel MERGE on cost grounds, and there is no documented way to force a parallel plan. There is an undocumented (and unsupported) way, using trace flag 8649:

SELECT
    row_count = SUM(Subtotals.cnt)
FROM @P AS p
CROSS APPLY
(
    SELECT
        cnt = COUNT_BIG(*)
    FROM dbo.test_transaction_item_detail AS i
    JOIN dbo.test_transaction_properties AS t ON
        t.transactionID = i.transactionID
    WHERE 
        $PARTITION.pf_test_transactionId(t.transactionID) = p.partition_number
        AND $PARTITION.pf_test_transactionId(i.transactionID) = p.partition_number
) AS SubTotals
OPTION (QUERYTRACEON 8649);

Now the query plan shows partition numbers from @P being distributed among threads on a round-robin basis. Each thread runs the inner side of the nested loops join for a single partition, achieving our goal of processing disjoint data concurrently. The same result is now returned in 3 seconds on my 8 hyper-cores, with all eight at 100% utilization.

I am not recommending you use this technique necessarily - see my earlier warnings - but it does address your question.

See my article Improving Partitioned Table Join Performance for more details.

Columnstore

Seeing as you are using SQL Server 2012 (and assuming it is Enterprise) you also have the option of using a columnstore index. This shows the potential of batch mode hash joins where sufficient memory is available:

CREATE NONCLUSTERED COLUMNSTORE INDEX cs 
ON dbo.test_transaction_properties (transactionID);

CREATE NONCLUSTERED COLUMNSTORE INDEX cs 
ON dbo.test_transaction_item_detail (transactionID);

With these indexes in place the query...

SELECT
    COUNT_BIG(*)
FROM dbo.test_transaction_properties AS ttp
JOIN dbo.test_transaction_item_detail AS ttid ON
    ttid.transactionID = ttp.transactionID;

...results in the following execution plan from the optimizer without any trickery:

Correct results in 2 seconds, but eliminating the row-mode processing for the scalar aggregate helps even more:

SELECT
    COUNT_BIG(*)
FROM dbo.test_transaction_properties AS ttp
JOIN dbo.test_transaction_item_detail AS ttid ON
    ttid.transactionID = ttp.transactionID
GROUP BY
    ttp.transactionID % 1;

The optimized column-store query runs in 851ms.

Geoff Patterson created the bug report Partition Wise Joins but it was closed as Won't Fix.

Answer 2

The way to make the optimizer work the way you think better is via query hints.

In this case, OPTION (MERGE JOIN)

Or you can go the whole hog and use USE PLAN