Частичная агрегация

 Автор оригинала: Craig Freedman

В нескольких предыдущих статьях обсуждалось, как в SQL Server реализована агрегация, были рассмотрены операторы агрегирования потока и хеш-агрегат. Я также использовал хеш-агрегат в качестве примера в статье Введение в распараллеливание исполнения запроса. В этой статье мы рассмотрим частичную агрегацию. Частичная агрегация — это метод, который SQL Server использует для оптимизации параллельной агрегации. Прежде чем начать, я хочу отметить, что рассуждения о частичной агрегацию можно найти в книге Inside Microsoft SQL Server 2005 : Query Tuning and Optimization (см. страницу 187 внизу).

Начнем с простого примера скалярной агрегации. Напомним, что скалярный агрегат — это агрегат без предложения GROUP BY. Скалярный агрегат всегда выдаёт одну строку.

CREATE TABLE T (A INT, B INT IDENTITY, C INT, D INT) 
CREATE CLUSTERED INDEX TA ON T(A)

SELECT COUNT(*) FROM T

Этот запрос предсказуемо дает тривиальный план со Stream Aggregate:

  |--Compute Scalar(DEFINE:([Expr1004]=CONVERT_IMPLICIT(int,[Expr1005],0)))
       |--Stream Aggregate(DEFINE:([Expr1005]=Count(*)))
            |--Clustered Index Scan(OBJECT:([T].[TA]))

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

В качестве альтернативы можно было бы организовать распараллеливание потоков данных между операторами Stream Aggregate и операторами Clustered Index Scan. Это позволит использовать распараллеленный просмотр, причём вычисления будут делаться в одном потоке и вернут одну строку. Однако в таком случае каждая строка из просмотра передавалась бы оператору Exchange, что повисит накладные расходы. Таким образом, этот вариант, хотя и допустим, но не даст повышения производительности, чего хотелось бы добиться.

К счастью, есть третий вариант. Мы можем использовать распараллеленный просмотр, разделив вычислительную задачу на несколько потоков (как в первом варианте), использовав оператор Exchange для объединения подсчётов по каждому потоку в один поток и, наконец, суммировать вычисления в каждом потоке, чтобы получить общее число. Эта стратегия более эффективна, поскольку нам нужно передать оператору Exchange только по одной строке каждого потока. Чтобы оптимизатор использовал такой план, нам нужно добавить в таблицу побольше данных. Но, чтобы сэкономить время, давайте используем UPDATE STATISTICS, для того, чтобы заставить оптимизатор подумать, что в таблице много строк:

UPDATE STATISTICS T WITH ROWCOUNT = 1000000, PAGECOUNT = 100000

SELECT COUNT(*) FROM T OPTION (RECOMPILE)

Потребуется добавить оптимизатору подсказку RECOMPILE, чтобы он создал план запроса с новой статистикой. Вот какой план должен получится:

  |--Compute Scalar(DEFINE:([Expr1004]=CONVERT_IMPLICIT(int,[globalagg1006],0)))
       |--Stream Aggregate(DEFINE:([globalagg1006]=SUM([partialagg1005])))
            |--Parallelism(Gather Streams)
                 |--Stream Aggregate(DEFINE:([partialagg1005]=Count(*)))
                      |--Clustered Index Scan(OBJECT:([T].[TA]))

Самый нижний оператор агрегирования называется частичным агрегатом, поскольку он вычисляет только часть результата. Иногда его называют локальным агрегатом, поскольку его часть результата является локальной для потока, в котором он выполняется. Самый верхний агрегат называется глобальным агрегатом, поскольку он вычисляет окончательный результат.

SQL Server может использовать частичную агрегацию для большинства агрегатных функций, включая такие, как: COUNT, SUM, AVG, MIN и MAX. Хотя частичная агрегация необходима для распараллеливания скалярных агрегатов, она также полезна и для агрегатов с предложением GROUP BY. Решение оптимизатора использовать частичную агрегацию зависит от количества уникальных групп и размера этих групп. Если оптимизатор ожидает, что запрос сгенерирует несколько больших групп (например, в случае скалярной агрегации), он будет использовать частичную агрегацию. Однако если оптимизатор ожидает, что запрос создаст несколько небольших групп, он может выбрать использование одноуровневой агрегации. В случае небольших групп частичное агрегирование не может существенно снизить количество строк и просто увеличивает нагрузку при исполнении запроса. Более того, при наличии большого количества групп агрегацию легко распараллелить, хешируя ключи GROUP BY и распределяя разные группы по разным потокам.

Давайте посмотрим, как оптимизатор делает этот выбор. Колонка B в нашем примере имеет свойство IDENTITY. Хотя у нас нет реальных данных, этого свойства достаточно, чтобы заставить оптимизатора прийти к выводу, что этот столбец по большей части уникален (без уникального индекса оптимизатор не может быть уверен, что столбец действительно уникален, и будет предполагать, что это не так). Предположим, мы агрегируем по этому столбцу:

SELECT COUNT(*) FROM T GROUP BY B

  |--Parallelism(Gather Streams)
       |--Compute Scalar(DEFINE:([Expr1004]=CONVERT_IMPLICIT(int,[Expr1007],0)))
            |--Hash Match(Aggregate, HASH:([T].[B]) DEFINE:([Expr1007]=COUNT(*)))
                 |--Parallelism(Repartition Streams, Hash Partitioning, PARTITION COLUMNS:([T].[B]))
                      |--Clustered Index Scan(OBJECT:([T].[TA]))

Обратите внимание, что этот запрос использует обычный одноуровневый, хотя и с параллелизмом, оператор агрегации. Теперь предположим, что мы агрегируем данные по столбцу C, который не имеет свойства IDENTITY:

SELECT COUNT(*) FROM T GROUP BY C

  |--Compute Scalar(DEFINE:([Expr1004]=CONVERT_IMPLICIT(int,[globalagg1006],0)))
       |--Parallelism(Gather Streams)
            |--Stream Aggregate(GROUP BY:([T].[C]) DEFINE:([globalagg1006]=SUM([partialagg1005])))
                 |--Sort(ORDER BY:([T].[C] ASC))
                      |--Parallelism(Repartition Streams, Hash Partitioning, PARTITION COLUMNS:([T].[C]))
                           |--Hash Match(Partial Aggregate, HASH:([T].[C]), RESIDUAL:([T].[C] = [T].[C]) DEFINE:([partialagg1005]=COUNT(*)))
                                |--Clustered Index Scan(OBJECT:([T].[TA]))

На этот раз мы видим в плане Partial Aggregate. Также обратите внимание, что частичный агрегат представляет собой агрегацию HASH, а глобальный агрегат — Stream Aggregate. Оптимизатор в плане с частичным агрегированием может выбрать для частичных и глобальных агрегатов любой оператор физической агрегации (поток или хэш). Решение о том, какой оператор использовать, зависит от затрат.

Наконец, стоит отметить, что, хотя агрегат потока ведет себя одинаково независимо от
того, вычисляет ли он частичный или глобальный агрегат, частичный хэш-агрегат
немного отличается от обычного. Во-первых, частичный хэш-агрегат запрашивает только фиксированный минимальный объем памяти, поскольку предполагает, что он будет вычислять относительно небольшое количество групп. Во-вторых, он никогда не материализует строки в базе данных tempdb. Если частичному хеш-агрегату не хватает памяти, он просто прекращает агрегацию и начинает возвращать не агрегированные строки. Такое поведение безопасно, поскольку глобальный агрегат всегда будет считать правильно, независимо от того, что делает частичный агрегат. Частичный агрегат — это просто оптимизация производительности, цель которой — сократить расходование ресурсов глобальным агрегатом.