Colocation Join

Colocation Join — это возможность оптимизации Join в Doris. Она размещает данные нескольких таблиц на одних и тех же узлах BE по одинаковым правилам, что позволяет выполнять операции Join по столбцам бакетирования локально, избегая передачи данных между узлами и ускоряя запросы.

В этом документе описаны принципы работы, реализация, использование и особенности Colocation Join.

Проверка применимости

Перед использованием Colocation Join убедитесь, что выполняются следующие условия:

• Две или более таблицы, участвующие в Join, добавлены в одну Colocation Group.
• Ключ Join (Join Key) совпадает со столбцом бакетирования (Distribution Key).
• У таблиц одинаковое количество реплик и бакетов, а распределение данных стабильно (IsStable = true).
• В запросе есть явное узкое место производительности: Shuffle при Join больших таблиц.

Терминология

Термин	Сокращение	Описание
Colocation Group	CG	CG содержит одну или несколько таблиц. Таблицы одной группы имеют общую Colocation Group Schema и одинаковое распределение сегментов данных.
Colocation Group Schema	CGS	Описывает общие сведения о схеме, связанные с Colocation, для таблиц в CG: типы столбцов бакетирования, количество бакетов и количество реплик.

Принципы работы

Colocation Join объединяет таблицы с одинаковой CGS в одну CG и обеспечивает размещение соответствующих сегментов данных этих таблиц на одних и тех же узлах BE. Благодаря этому операции Join по столбцам бакетирования для таблиц в CG можно выполнять непосредственно над локальными данными, сокращая время передачи данных между узлами.

Бакеты и BucketsSequence

Данные таблицы распределяются по бакетам путем хеширования значений столбцов бакетирования и вычисления остатка от деления на количество бакетов. Предположим, у таблицы 8 бакетов. Тогда всего имеется 8 бакетов: [0, 1, 2, 3, 4, 5, 6, 7]. Эта последовательность называется BucketsSequence. Каждый бакет содержит один или несколько сегментов данных (Tablet):

• В таблице с одной партицией бакет содержит только один Tablet.
• В таблице с несколькими партициями бакет содержит несколько Tablet.

Ограничения для таблиц в одной CG

Чтобы обеспечить одинаковое распределение данных, у таблиц в одной CG должны совпадать следующие свойства:

1. Столбцы бакетирования и количество бакетов

   Столбцы бакетирования задаются в DISTRIBUTED BY HASH(col1, col2, ...) в операторе создания таблицы. Они определяют, значения каких столбцов используются для хеширования и распределения данных таблицы по разным Tablet. У таблиц в одной CG должны в точности совпадать типы и количество столбцов бакетирования, а также количество бакетов, чтобы сегменты данных нескольких таблиц можно было распределить с взаимно однозначным соответствием.

2. Количество реплик

   Количество реплик должно быть одинаковым для всех партиций всех таблиц в одной CG. Иначе для реплики какого-либо Tablet на том же BE может не оказаться соответствующей реплики сегмента другой таблицы.

Примечание

У таблиц в одной CG не обязательно должны совпадать количество партиций, диапазоны партиций или типы столбцов партиционирования.

Пример распределения данных

После задания столбцов бакетирования и количества бакетов таблицы в одной CG используют общую BucketsSequence. Количество реплик определяет, на каких узлах BE хранятся реплики Tablet из каждого бакета.

Предположим, BucketsSequence имеет вид [0, 1, 2, 3, 4, 5, 6, 7] и имеется 4 узла BE [A, B, C, D]. Один из возможных вариантов распределения данных:

+---+ +---+ +---+ +---+ +---+ +---+ +---+ +---+
| 0 | | 1 | | 2 | | 3 | | 4 | | 5 | | 6 | | 7 |
+---+ +---+ +---+ +---+ +---+ +---+ +---+ +---+
| A | | B | | C | | D | | A | | B | | C | | D |
|   | |   | |   | |   | |   | |   | |   | |   |
| B | | C | | D | | A | | B | | C | | D | | A |
|   | |   | |   | |   | |   | |   | |   | |   |
| C | | D | | A | | B | | C | | D | | A | | B |
+---+ +---+ +---+ +---+ +---+ +---+ +---+ +---+

Данные всех таблиц в CG распределяются единообразно по приведенному выше правилу. Это гарантирует, что данные с одинаковыми значениями столбцов бакетирования находятся на одном узле BE, позволяя выполнять Join локально.

Использование

Указание Colocation Group при создании таблицы

Назначение: добавить создаваемую таблицу в указанную Colocation Group.

Команда: укажите "colocate_with" = "group_name" в PROPERTIES.

Пример:

CREATE TABLE tbl (k1 int, v1 int sum)
DISTRIBUTED BY HASH(k1)
BUCKETS 8
PROPERTIES(
    "colocate_with" = "group1"
);

Описание:

• Если указанная группа не существует, Doris автоматически создает группу, содержащую только текущую таблицу.
• Если группа уже существует, Doris проверяет, соответствует ли текущая таблица Colocation Group Schema. Если соответствует, таблица создается и добавляется в группу. Сегменты и реплики таблицы создаются по правилам распределения данных существующей группы.
• Группа принадлежит базе данных, и ее имя уникально в пределах этой базы данных. Во внутреннем хранилище полное имя группы имеет вид dbId_groupName, но пользователи видят только groupName.

Создание глобальной группы для разных баз данных

Совет

В версии 2.0 Doris поддерживает группы, объединяющие таблицы из разных баз данных.

Назначение: выполнять Colocate Join между таблицами из разных баз данных.

Команда: при создании таблицы используйте ключевое слово __global__ в качестве префикса имени группы.

Пример:

CREATE TABLE tbl (k1 int, v1 int sum)
DISTRIBUTED BY HASH(k1)
BUCKETS 8
PROPERTIES(
    "colocate_with" = "__global__group1"
);

Описание: группа с префиксом __global__ не принадлежит отдельной базе данных, а ее имя уникально глобально. Создание глобальной группы позволяет выполнять Colocate Join между таблицами из разных баз данных.

Удаление таблицы

После полного удаления последней таблицы из группы сама группа также удаляется автоматически.

Примечание

Полное удаление означает удаление из корзины. Обычно после выполнения команды DROP TABLE таблица по умолчанию остается в корзине в течение одного дня, после чего удаляется.

Просмотр групп

Назначение: просмотреть сведения о существующих Colocation Group в кластере и распределении их данных.

1. Просмотр всех групп в кластере

SHOW PROC '/colocation_group';

+-------------+--------------+--------------+------------+----------------+----------+----------+
| GroupId     | GroupName    | TableIds     | BucketsNum | ReplicationNum | DistCols | IsStable |
+-------------+--------------+--------------+------------+----------------+----------+----------+
| 10005.10008 | 10005_group1 | 10007, 10040 | 10         | 3              | int(11)  | true     |
+-------------+--------------+--------------+------------+----------------+----------+----------+

Описание полей:

Поле	Описание
GroupId	Уникальный в пределах кластера идентификатор группы. Первая часть — идентификатор базы данных, вторая — идентификатор группы.
GroupName	Полное имя группы.
TableIds	Список идентификаторов таблиц, входящих в группу.
BucketsNum	Количество бакетов.
ReplicationNum	Количество реплик.
DistCols	Столбцы распределения, то есть типы столбцов бакетирования.
IsStable	Стабильна ли группа (определение стабильного состояния см. в разделе Балансировка и восстановление реплик Colocation).

2. Просмотр распределения данных конкретной группы

SHOW PROC '/colocation_group/10005.10008';

+-------------+---------------------+
| BucketIndex | BackendIds          |
+-------------+---------------------+
| 0           | 10004, 10002, 10001 |
| 1           | 10003, 10002, 10004 |
| 2           | 10002, 10004, 10001 |
| 3           | 10003, 10002, 10004 |
| 4           | 10002, 10004, 10003 |
| 5           | 10003, 10002, 10001 |
| 6           | 10003, 10004, 10001 |
| 7           | 10003, 10004, 10002 |
+-------------+---------------------+

Описание полей:

Поле	Описание
BucketIndex	Индекс в последовательности бакетов.
BackendIds	Список идентификаторов узлов BE, на которых находятся сегменты данных бакета.

Примечание

Приведенные выше команды требуют привилегии ADMIN и недоступны обычным пользователям.

Изменение свойства Colocate Group у таблицы

Назначение: добавить существующую таблицу в Colocation Group, переместить ее в другую группу или удалить из группы.

1. Назначение или смена группы

ALTER TABLE tbl SET ("colocate_with" = "group2");

Поведение:

• Если таблица ранее не принадлежала ни одной группе, команда проверяет схему и добавляет таблицу в группу (если группа не существует, она создается).
• Если таблица ранее принадлежала другой группе, команда сначала удаляет таблицу из исходной группы, а затем добавляет в новую (если группа не существует, она создается).

2. Удаление свойства Colocation

ALTER TABLE tbl SET ("colocate_with" = "");

Другие связанные операции

При добавлении партиций (ADD PARTITION) или изменении количества реплик таблицы со свойством Colocation Doris проверяет, не нарушает ли изменение Colocation Group Schema. Если нарушает, изменение отклоняется.

Балансировка и восстановление реплик Colocation

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

Стабильное состояние группы

У самой группы есть свойство Stable:

Состояние	Значение	Влияние на запросы
Stable (true)	Ни один из сегментов таблиц группы в данный момент не изменяется.	Возможность Colocation работает в обычном режиме.
Unstable (false)	Некоторые сегменты таблиц группы восстанавливаются или перемещаются.	Colocation Join для соответствующих таблиц заменяется обычным Join.

Восстановление реплик

Реплики могут храниться только на заданных узлах BE. Поэтому, когда BE становится недоступен, например из-за сбоя или вывода из эксплуатации (Decommission), необходимо найти другой BE для замены. Doris предпочитает BE с наименьшей нагрузкой. После замены необходимо восстановить все сегменты данных бакета, находившиеся на прежнем BE. На время перемещения группа помечается как Unstable.

Балансировка реплик

Doris стремится равномерно распределять сегменты таблиц Colocation по всем узлам BE. Различия между двумя типами балансировки приведены ниже:

Тип	Единица балансировки	Описание
Обычная таблица	Отдельная реплика	BE с меньшей нагрузкой подбирается отдельно для каждой реплики.
Таблица Colocation	Бакет	Все реплики в бакете перемещаются вместе.

Используется простой алгоритм балансировки: фактический размер реплик не учитывается, а для равномерного распределения BucketsSequence по всем узлам BE используется только количество реплик. Подробности алгоритма см. в комментариях к коду в ColocateTableBalancer.java.

Примечание

• Примечание 1: текущий алгоритм балансировки и восстановления реплик Colocation может плохо работать в неоднородных кластерах Doris. Неоднородность означает, что узлы BE различаются по емкости, количеству или типу дисков (SSD и HDD). В таких кластерах узлы BE с небольшой и большой емкостью могут в итоге хранить одинаковое количество реплик.
• Примечание 2: когда группа находится в состоянии Unstable, операции Join с ее таблицами заменяются обычными Join. Это может значительно снизить производительность запросов в кластере. Если автоматическая балансировка нежелательна, ее можно отключить с помощью параметра конфигурации FE disable_colocate_balance и снова включить в подходящий момент (подробнее см. в разделе Расширенные операции).

Выполнение запросов

Запросы к таблицам Colocation выполняются так же, как к обычным таблицам: пользователям не нужно учитывать свойство Colocation. Если группа таблицы Colocation находится в состоянии Unstable, запрос автоматически переходит на обычный Join.

Следующий пример показывает, как проверить, используется ли Colocation Join.

Примеры таблиц

Таблица 1:

CREATE TABLE `tbl1` (
    `k1` date NOT NULL COMMENT "",
    `k2` int(11) NOT NULL COMMENT "",
    `v1` int(11) SUM NOT NULL COMMENT ""
) ENGINE=OLAP
AGGREGATE KEY(`k1`, `k2`)
PARTITION BY RANGE(`k1`)
(
    PARTITION p1 VALUES LESS THAN ('2019-05-31'),
    PARTITION p2 VALUES LESS THAN ('2019-06-30')
)
DISTRIBUTED BY HASH(`k2`) BUCKETS 8
PROPERTIES (
    "colocate_with" = "group1"
);

Таблица 2:

CREATE TABLE `tbl2` (
    `k1` datetime NOT NULL COMMENT "",
    `k2` int(11) NOT NULL COMMENT "",
    `v1` double SUM NOT NULL COMMENT ""
) ENGINE=OLAP
AGGREGATE KEY(`k1`, `k2`)
DISTRIBUTED BY HASH(`k2`) BUCKETS 8
PROPERTIES (
    "colocate_with" = "group1"
);

Просмотр плана запроса

DESC SELECT * FROM tbl1 INNER JOIN tbl2 ON (tbl1.k2 = tbl2.k2);

+----------------------------------------------------+
| Explain String                                     |
+----------------------------------------------------+
| PLAN FRAGMENT 0                                    |
|  OUTPUT EXPRS:`tbl1`.`k1` |                        |
|   PARTITION: RANDOM                                |
|                                                    |
|   RESULT SINK                                      |
|                                                    |
|   2:HASH JOIN                                      |
|   |  join op: INNER JOIN                           |
|   |  hash predicates:                              |
|   |  colocate: true                                |
|   |    `tbl1`.`k2` = `tbl2`.`k2`                   |
|   |  tuple ids: 0 1                                |
|   |                                                |
|   |----1:OlapScanNode                              |
|   |       TABLE: tbl2                              |
|   |       PREAGGREGATION: OFF. Reason: null        |
|   |       partitions=0/1                           |
|   |       rollup: null                             |
|   |       buckets=0/0                              |
|   |       cardinality=-1                           |
|   |       avgRowSize=0.0                           |
|   |       numNodes=0                               |
|   |       tuple ids: 1                             |
|   |                                                |
|   0:OlapScanNode                                   |
|      TABLE: tbl1                                   |
|      PREAGGREGATION: OFF. Reason: No AggregateInfo |
|      partitions=0/2                                |
|      rollup: null                                  |
|      buckets=0/0                                   |
|      cardinality=-1                                |
|      avgRowSize=0.0                                |
|      numNodes=0                                    |
|      tuple ids: 0                                  |
+----------------------------------------------------+

Если Colocation Join используется, в узле Hash Join отображается colocate: true.

Если он не используется, план запроса выглядит следующим образом:

+----------------------------------------------------+
| Explain String                                     |
+----------------------------------------------------+
| PLAN FRAGMENT 0                                    |
|  OUTPUT EXPRS:`tbl1`.`k1` |                        |
|   PARTITION: RANDOM                                |
|                                                    |
|   RESULT SINK                                      |
|                                                    |
|   2:HASH JOIN                                      |
|   |  join op: INNER JOIN (BROADCAST)               |
|   |  hash predicates:                              |
|   |  colocate: false, reason: group is not stable  |
|   |    `tbl1`.`k2` = `tbl2`.`k2`                   |
|   |  tuple ids: 0 1                                |
|   |                                                |
|   |----3:EXCHANGE                                  |
|   |       tuple ids: 1                             |
|   |                                                |
|   0:OlapScanNode                                   |
|      TABLE: tbl1                                   |
|      PREAGGREGATION: OFF. Reason: No AggregateInfo |
|      partitions=0/2                                |
|      rollup: null                                  |
|      buckets=0/0                                   |
|      cardinality=-1                                |
|      avgRowSize=0.0                                |
|      numNodes=0                                    |
|                                                    |
| PLAN FRAGMENT 1                                    |
|  OUTPUT EXPRS:                                     |
|   PARTITION: RANDOM                                |
|                                                    |
|   STREAM DATA SINK                                 |
|     EXCHANGE ID: 03                                |
|     UNPARTITIONED                                  |
|                                                    |
|   1:OlapScanNode                                   |
|      TABLE: tbl2                                   |
|      PREAGGREGATION: OFF. Reason: null             |
|      partitions=0/1                                |
|      rollup: null                                  |
|      buckets=0/0                                   |
|      cardinality=-1                                |
|      avgRowSize=0.0                                |
|      numNodes=0                                    |
|      tuple ids: 1                                  |
+----------------------------------------------------+

В узле HASH JOIN отображается соответствующая причина: colocate: false, reason: group is not stable, а в плане появляется узел EXCHANGE.

Сравнение типов Join

Чтобы помочь определить, используется ли в запросе Colocation Join, в следующей таблице сравниваются распространенные типы Join в Doris:

Тип Join	Перераспределение данных (Shuffle)?	Условие применения
Colocate Join	Нет	Таблицы входят в одну Colocate Group и IsStable=true.
Bucket Shuffle Join	Частичное (с одной стороны)	Ключ Join совпадает со столбцом бакетирования левой таблицы.
Shuffle Join	Да (с обеих сторон)	Поведение по умолчанию, если ни одно из перечисленных выше условий не выполняется.
Broadcast Join	Да (небольшая таблица)	Правая таблица имеет небольшой размер.

Расширенные операции

Параметры конфигурации FE

Параметр	Значение по умолчанию	Описание
disable_colocate_relocate	false	Отключать ли автоматическое восстановление реплик Colocation в Doris. Значение по умолчанию — false (не отключено). Параметр влияет только на восстановление реплик таблиц Colocation и не затрагивает обычные таблицы.
disable_colocate_balance	false	Отключать ли автоматическую балансировку реплик Colocation в Doris. Значение по умолчанию — false (не отключено). Параметр влияет только на балансировку реплик таблиц Colocation и не затрагивает обычные таблицы.
disable_colocate_join	См. описание	Отключать ли Colocation Join. В версиях 0.10 и более ранних значение по умолчанию — true (отключено). В одной из последующих версий значением по умолчанию станет false (включено).
use_new_tablet_scheduler	См. описание	В версиях 0.10 и более ранних новая логика планирования реплик несовместима с Colocation Join. Поэтому в версиях 0.10 и более ранних при disable_colocate_join = false необходимо задать use_new_tablet_scheduler = false, чтобы отключить новый планировщик реплик. В последующих версиях use_new_tablet_scheduler всегда имеет значение true.

Совет

Параметры disable_colocate_relocate и disable_colocate_balance, указанные выше, можно изменять динамически. Подробнее см. HELP SHOW CONFIG; и HELP SET CONFIG;.

HTTP Restful API

Doris предоставляет несколько HTTP Restful API, связанных с Colocation Join, для просмотра и изменения Colocation Group.

Эти API реализованы на стороне FE и доступны по адресу fe_host:fe_http_port. Для их использования требуется привилегия ADMIN.

1. Просмотр всех сведений о Colocation в кластере

GET /api/colocate

Returns the internal Colocation information in JSON format.

{
    "msg": "success",
    "code": 0,
    "data": {
        "infos": [
            ["10003.12002", "10003_group1", "10037, 10043", "1", "1", "int(11)", "true"]
        ],
        "unstableGroupIds": [],
        "allGroupIds": [{
            "dbId": 10003,
            "grpId": 12002
        }]
    },
    "count": 0
}

2. Установка состояния Stable или Unstable для группы

• Установить Stable

  DELETE /api/colocate/group_stable?db_id=10005&group_id=10008
  
  Returns: 200

• Установить Unstable

  POST /api/colocate/group_stable?db_id=10005&group_id=10008
  
  Returns: 200

3. Задание распределения данных группы

Этот метод позволяет принудительно задать распределение данных группы.

POST /api/colocate/bucketseq?db_id=10005&group_id=10008

Body:
[[10004,10002],[10003,10002],[10002,10004],[10003,10002],[10002,10004],[10003,10002],[10003,10004],[10003,10004],[10003,10004],[10002,10004]]

Returns 200

Тело запроса (Body) представляет собой вложенный массив, задающий BucketsSequence и идентификаторы узлов BE, по которым распределены сегменты каждого бакета.

Примечание

При использовании этой команды может потребоваться установить для параметров конфигурации FE disable_colocate_relocate и disable_colocate_balance значение true, то есть отключить автоматическое восстановление и балансировку реплик Colocation. Иначе система может автоматически отменить внесенное изменение.

Часто задаваемые вопросы

colocate: false, reason: group is not stable в плане запроса

Это означает, что группа сейчас находится в состоянии Unstable, возможно, из-за выполняющегося восстановления или балансировки реплик. В этом случае Join заменяется обычным Join. Когда группа вернется в состояние Stable, Colocation Join снова можно будет использовать. Проверить поле IsStable можно с помощью SHOW PROC '/colocation_group';.

Как проверить, доступна ли Colocation Group в данный момент

Выполните SHOW PROC "/colocation_group"; и проверьте поле IsStable. Значение true означает, что группа доступна и Join может использовать план Colocate; false означает, что группа временно недоступна, пока Doris выполняет балансировку данных.

Как долго сохраняется состояние IsStable=false

Это зависит от объема перемещаемых данных и нагрузки на кластер. Состояние восстанавливается автоматически после завершения балансировки Tablet в Doris. Если значение false сохраняется долго, обратитесь к приведенным ниже рекомендациям по устранению неполадок.

Ошибка при создании таблицы: таблица не может войти в указанную группу

Проверьте, выполняются ли все следующие условия:

• Типы и количество столбцов бакетирования в точности совпадают с типами и количеством таких столбцов у существующих таблиц группы.
• Количество бакетов совпадает.
• Количество реплик одинаково для всех партиций.

Любое несоответствие не позволяет добавить таблицу в группу.

Группа долго остается в состоянии Unstable

Возможные причины:

• В кластере произошел сбой BE или выполняется Decommission, а восстановление реплик еще не завершено.
• Неоднородность кластера затрудняет достижение равномерного распределения при балансировке.
• Автоматическая балансировка не отключена и постоянно запускает перемещение данных.

Можно временно отключить автоматическую балансировку, задав disable_colocate_balance = true, а после стабилизации кластера снова включить ее.

Могут ли две таблицы из разных баз данных выполнять Colocation Join

Да. В версии 2.0 и более поздних можно создавать таблицы, указывая имя глобальной группы с префиксом __global__.