По материалам статьи Craig Freedman: Parallel Hash Join

Перевод Ирины Наумовой.

SQL Server использует один из двух вариантов стратегии распараллеливания хэш-соединения. Наиболее часто встречается хэш-секционирование (Hash Partitioning). Реже можно встретить Broadcast-секционирование; эту стратегию часто называют "Broadcast hash join".

Хэш-секционирование

Типичное распараллеливания хеш-соединения включает распределение строк компановки (т.е. строк из первого входого потока) и пробных строк (т.е. строк из второго входного потока) между отдельными потоками хэш-соединения с использованием хэш-секционирования. Если строки компановки и пробы имеют одни и те же значения ключа (т.е. они участвуют в соединении), такие хэши гарантированно попадут в один и тот же поток хэш-соединения. После того как данные были секционированы по хэшам между потоками, все экземпляры хэш-соединений работают полностью независимо от соответствующих им наборов данных. Отсутствие любых межпоточных зависимостей гарантирует, что эта стратегия будет давать очень хорошую масштабируемость по мере повышения степени параллелизма (т.е. числа потоков).
Как и в других рассмотренных ранее примерах с параллелизмом, в этой статье используется достаточно большая таблица, что бы, работая с ней, оптимизатор выбирал параллельный план исполнения запроса. Если вы будете воспроизводить эти примеры у себя, учтите, что создание этой таблицы может затянуться на несколько минут.

create table T1 (a int, b int, x char(200))
set nocount on
declare @i int
set @i = 0
while @i < 1000000
  begin
    insert T1 values(@i, @i, @i)
    set @i = @i + 1
  end
select * into T2 from T1
select * into T3 from T1
select * from T1 join T2 on T1.b = T2.a 

  ^{|--Parallelism(Gather Streams)
       |--Hash Match(Inner Join, HASH:([T1].[b])=([T2].[a]), RESIDUAL:([T2].[a]=[T1].[b]))
            |--Parallelism(Repartition Streams, Hash Partitioning, PARTITION COLUMNS:([T1].[b]))
            |    |--Table Scan(OBJECT:([T1]))
            |--Parallelism(Repartition Streams, Hash Partitioning, PARTITION COLUMNS:([T2].[a]))
                 |--Table Scan(OBJECT:([T2]))}

Обратите внимание, что в отличие от Nested Loops, у нас есть по одному оператору Exchange (оператор параллелизма, обозначенный как "Parallelism") для каждой ветки плана со сканированием таблиц под хэш-соединением (оба входных потока: компановка и проба). Тут Exchange используется для распределения хэшей по разным потокам хэш-соединения.

Broadcast-секционирование

Рассмотрим, что произойдет, если мы попытаемся распараллелить хэш-соединение с использованием хэш-секционирования, когда у нас не много строк в ветке компановки хэш-соединения. Если строк меньше, чем потоков хэш-соединения, некоторые потоки вообще не смогут получить ни одной строки. В таком случае, такие потоки в фазе пробы показанного выше соединения останутся без работы и будут бездействовать. Даже если строк будет больше, чем потоков, в связи с наличием дубликатов значений ключа и/или перекосов в хэш-функции, одни потоки могут получить намного больше строк, чем другие.
Чтобы исключить риск подобных перекосов, когда оптимизатор считает число строк компановки относительно небольшим, он может выбрать для передачи этих строк все потоки хэш-соединения. Так как все строки компановки транслируются на все хэш-соединения потока, в Broadcast хэш-соединении не имеет значения, куда будут направлены строки пробы. Каждая строка пробы может быть отправлена в любой поток и, если она может участововать в соединении с какой - либо строкой компановки, так и будет.

Вот пример:

select * from T1 join T2 on T1.b = T2.a where T1.a = 0

  ^{|--Parallelism(Gather Streams)
       |--Hash Match(Inner Join, HASH:([T1].[b])=([T2].[a]), RESIDUAL:([T2].[a]=[T1].[b]))
            |--Parallelism(Distribute Streams, Broadcast Partitioning)
            |    |--Table Scan(OBJECT:([T1]), WHERE:([T1].[a]=(0)))
            |--Table Scan(OBJECT:([T2]))}

Обратите внимание, что оператор параллелизма расположенный выше оператора простмотра T1 теперь превратился в Broadcast, в то время как полностью исчезло распараллеливание просмотра T2. Нам не нужен параллелизм выше T2, потому что Распараллеленный Просмотр автоматически распределяет страницы и строки T2 потоками хэш-соединения. Это подобно тому, как Распараллеленный Просмотр распределяет строки между потоками Nested Loops Join в случае Parallel Nested Loops Join.
Как и в распараллеленном соединении вложенных циклов, если на входе пробы Broadcast хэш-соединения встретится серия значений  (например, какэто бывает, если присутствует оператор TOP), тогда, возможно, для распараллеливания потребуется перераспределение строк в порядке круговой очереди.

И так, если Broadcast хэш-соединениe настолько хороши (они снижают риски перекосов), то почему бы нам не использовать их везде? Закавыка в том, что Broadcast хэш-соединениe использует больше оперативной памяти, чем его коллега хэш-секционирование. Это происходит, поскольку мы предоставляем каждую строку сборки для каждого потока хэш-соединения, и, если использовать вдвое больше потоков, то удвоится и объем нужной для строк сборки памяти. При распараллеленном хэш-соединении с хэш-секционированием будет использоваться одинаковый объём памяти, независимо от степени параллелизма.

Bitmap фильтрация

  ^{|--Parallelism(Gather Streams)
       |--Hash Match(Inner Join, HASH:([T1].[b])=([T2].[a]), RESIDUAL:([T2].[a]=[T1].[b]))
            |--Bitmap(HASH:([T1].[b]), DEFINE:([Bitmap1008]))
            |    |--Parallelism(Repartition Streams, Hash Partitioning, PARTITION COLUMNS:([T1].[b]))
            |         |--Table Scan(OBJECT:([T1]), WHERE:([T1].[a]<(100000)))
            |--Parallelism(Repartition Streams, Hash Partitioning, PARTITION COLUMNS:([T2].[a]), WHERE:(PROBE([Bitmap1008])=TRUE))
                 |--Table Scan(OBJECT:([T2]))}

Что представляет собой оператор Bitmap? Предикат "T1.a < 100000" убирает 90% строк компановки из T1. Он также косвенно убирает 90% строк в выборке из таблицы T2, потому что они не участвуют в соединении со строками из T1. Оператор Bitmap обеспечивает эффективный способ фильтрации T1 непосредственно по T2 без необходимости протаскивания строк по всему пути через оператор параллелизма к соединению. Как следует из названия, этот оператор создаёт битовую карту. Так же, как и в хэш-соединении, берётся хэш каждой строки T1 по ключу соединения T1.b и устанавливается соответствующий бит в битовой карте. По окончании просмотра Т1 будет завершено и хэш-соединение сборки, битовая карта будет передана оператору параллелизма (который видно на плане выше просмотра Т2) где карта используется в качестве фильтра. Тут берётся хэш каждой строки T2 по ключу соединения T2.a и проверяется соответствующий ему бит в битовой карте. Если бит установлен, то строка участвует в соединении и проходит дальше к хэш-соединению. Если бит не установлен, то строка не участвует в соединении и будет отброшена. Более подробную информацию о битовых масках смотрите в этой статье блога SQL Server Query Processing Team