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)