Manticore позволяет создавать распределённые таблицы, которые работают как обычные plain или real-time таблицы, но на самом деле являются коллекцией дочерних таблиц, используемых для поиска. Когда запрос отправляется в распределённую таблицу, он распределяется по всем таблицам в коллекции. Затем сервер собирает и обрабатывает ответы для сортировки и при необходимости перерасчёта значений агрегатов.

С точки зрения клиента, создаётся впечатление, что он выполняет запрос к одной таблице.

table foo {
    type = distributed
    local = bar
    local = bar1, bar2
    agent = 127.0.0.1:9312:baz
    agent = host1|host2:tbl
    agent = host1:9301:tbl1|host2:tbl2 [ha_strategy=random retry_count=10]
    ...
}

CREATE TABLE distributed_table type='distributed'
  local='local_table'
  agent='127.0.0.1:9312:remote_table';

Суть распределённой таблицы заключается в списке дочерних таблиц, на которые она ссылается. Существует два типа дочерних таблиц в распределённой таблице:

1. Локальные таблицы: Это таблицы, обслуживаемые на том же сервере, что и распределённая таблица. Для перечисления локальных таблиц используется синтаксис local =. Вы можете указать несколько локальных таблиц с помощью нескольких строк local = или объединить их в один список, разделённый запятыми.

2. Удалённые таблицы: Это таблицы, обслуживаемые вне сервера. Для перечисления удалённых таблиц используется синтаксис agent =. Каждая строка представляет один endpoint или агент. Каждый агент может иметь несколько внешних локаций и опций того, как он должен работать. Подробнее здесь. Важно отметить, что сервер не имеет информации о типе таблицы, с которой он работает. Это может привести к ошибкам, если, например, вы выполните CALL PQ для удалённой таблицы 'foo', которая не является percolate таблицей.

Удалённая таблица в Manticore Search представлена префиксом agent в определении распределённой таблицы. Распределённая таблица может включать сочетание локальных и удалённых таблиц. Если локальные таблицы не указаны, распределённая таблица будет полностью удалённой и будет выступать только как прокси. Например, у вас может быть экземпляр Manticore, который слушает несколько портов и обслуживает разные протоколы, а затем перенаправляет запросы на серверы backend, которые принимают соединения только через внутренний бинарный протокол Manticore, используя постоянные соединения для уменьшения накладных расходов при установлении соединений.
Хотя распределённая таблица, состоящая только из удалённых, сама не содержит локальных таблиц, она всё равно потребляет ресурсы машины, поскольку должна выполнять окончательные вычисления, такие как слияние результатов и вычисление итоговых агрегированных значений.

agent = address1 [ | address2 [...] ][:table-list]
agent = address1[:table-list [ | address2[:table-list [...] ] ] ]

Директива agent объявляет удаленные агенты, которые опрашиваются каждый раз при поиске по охватывающей распределенной таблице. Эти агенты, по сути, являются указателями на сетевые таблицы. Указанное значение включает адрес и также может включать несколько альтернатив (зеркала агента) либо только для адреса, либо для адреса и списка таблиц. См. Зеркалирование для синтаксиса зеркал и Балансировка нагрузки для информации о том, как выбираются зеркальные агенты.

Спецификация адреса должна быть одной из следующих:

address = hostname[:port] # eg. server2:9312
address = /absolute/unix/socket/path # eg. /var/run/manticore2.sock

hostname — это имя удаленного хоста, port — номер удаленного TCP-порта, table-list — это список имен таблиц, разделенных запятыми, а квадратные скобки [] обозначают необязательную часть. Для TCP-соединений этот порт является портом удаленного агента/API (обычно 9312), а не портом MySQL (9306).

Если имя таблицы опущено, предполагается, что это та же таблица, где определена эта строка. Другими словами, при определении агентов для распределённой таблицы 'mycoolindex' можно просто указать адрес, и будет предполагаться, что запрос идёт к таблице mycoolindex на конечных точках агента.

Если номер порта опущен, считается, что он равен 9312. Если он задан, но недопустим (например, 70000), агент будет пропущен.

Вы можете направлять каждого агента к одной или нескольким удалённым таблицам, расположенным на одном или нескольких сетевых серверах без ограничений. Это позволяет несколько различных режимов использования:

• Шардинг по нескольким серверам агентов и создание произвольной топологии кластера
• Шардирование по нескольким серверам-агентам с зеркалированием для обеспечения высокой доступности и балансировки нагрузки (см. Зеркалирование и Балансировка нагрузки)
• Шардинг внутри localhost для использования нескольких ядер (хотя проще использовать несколько локальных таблиц)

Чтобы избежать путаницы:

• Одна запись agent='host1|host2:table' означает один удаленный шард с зеркальными бэкендами.
• Несколько записей agent='...' означают несколько удаленных шардов.

Все агенты опрашиваются параллельно. Список индексов передаётся удалённому агенту без изменений. Точный способ поиска по этому списку в агенте (последовательно или параллельно) зависит только от конфигурации агента (см. настройку threads). Мастер не управляет этим удалённо.

Важно отметить, что опция LIMIT игнорируется в запросах к агентам. Это связано с тем, что каждый агент может содержать разные таблицы, и ответственность за применение ограничения к итоговому набору результатов лежит на клиенте. Поэтому запрос к физической таблице отличается от запроса к распределённой таблице, если смотреть в логах запросов. Запрос не может быть простой копией исходного запроса, так как это приведёт к некорректным результатам.

Например, если клиент делает запрос SELECT ... LIMIT 10, 10, а существует два агента, при этом второй агент содержит только 10 документов, трансляция исходного запроса LIMIT 10, 10 приведёт к тому, что от второго агента будут получены 0 документов. Однако запрос LIMIT 10,10 должен вернуть документы с 10 по 20 из итогового набора. Чтобы решить эту проблему, запрос отправляют агентам с более широким ограничением, например, используя значение max_matches по умолчанию 1000.

Например, если есть распределённая таблица dist, которая ссылается на удалённую таблицу user, запрос клиента SELECT * FROM dist LIMIT 10,10 будет преобразован в запрос SELECT * FROM user LIMIT 0,1000 и отправлен удалённой таблице user. После получения результата распределённая таблица применит LIMIT 10,10 и вернёт запрашиваемые 10 документов.

SELECT * FROM dist LIMIT 10,10;

запрос будет преобразован в:

SELECT * FROM user LIMIT 0,1000

Кроме того, значение может содержать опции для каждого отдельного агента, такие как:

• ha_strategy - random, roundrobin, nodeads, noerrors (переопределяет глобальную настройку ha_strategy для конкретного агента; см. также Зеркалирование)
• conn - pconn, persistent (эквивалентно установке agent_persistent на уровне таблицы)
• blackhole 0,1 (идентично настройке agent_blackhole для агента)
• retry_count целочисленное значение (соответствует agent_retry_count, но заданное значение не умножается на число зеркал)

agent = address1:table-list[[ha_strategy=value, conn=value, blackhole=value]]

Пример:

# config on box1
# sharding a table over 3 servers
agent = box2:9312:shard1
agent = box3:9312:shard2
# config on box2
# sharding a table over 3 servers
agent = box1:9312:shard2
agent = box3:9312:shard3
# config on box3
# sharding a table over 3 servers
agent = box1:9312:shard1
agent = box2:9312:shard3
# per agent options
agent = box1:9312:shard1[ha_strategy=nodeads]
agent = box2:9312:shard2[conn=pconn]
agent = box2:9312:shard2[conn=pconn,ha_strategy=nodeads]
agent = test:9312:any[blackhole=1]
agent = test:9312|box2:9312|box3:9312:any2[retry_count=2]
agent = test:9312|box2:9312:any2[retry_count=2,conn=pconn,ha_strategy=noerrors]

table products_dist {
  type  = distributed
  agent = 127.0.0.1:9312|127.0.0.1:9313:products
}

CREATE TABLE products_dist type='distributed'
  agent='127.0.0.1:9312:products|127.0.0.1:9313:products';

CREATE TABLE products_dist type='distributed'
  agent='127.0.0.1:9312:products|127.0.0.1:9313:products'
  ha_strategy='roundrobin'
  agent_connect_timeout='200ms'
  agent_query_timeout='500ms'
  mirror_retry_count='2';

SELECT id, title, node FROM products_dist;
SHOW CREATE TABLE products_dist;

+------+------------+-------+
| id   | title      | node  |
+------+------------+-------+
|    1 | same title | node1 |
+------+------------+-------+
+---------------+----------------------------------------------------------------------------------+
| Table         | Create Table                                                                     |
+---------------+----------------------------------------------------------------------------------+
| products_dist | CREATE TABLE products_dist type='distributed' agent='127.0.0.1:9312:products|... |
+---------------+----------------------------------------------------------------------------------+

Для оптимальной производительности рекомендуется помещать удалённые таблицы, расположенные на одном сервере, в одну запись. Например, вместо:

agent = remote:9312:idx1
agent = remote:9312:idx2

следует предпочесть:

agent = remote:9312:idx1,idx2

agent_persistent = remotebox:9312:index2

Опция agent_persistent позволяет устанавливать постоянное соединение с агентом, то есть соединение не закрывается после выполнения запроса. Синтаксис этой директивы такой же, как у директивы agent. Однако вместо открытия нового соединения с агентом для каждого запроса и последующего его закрытия мастер будет поддерживать открытое соединение и повторно использовать его для последующих запросов. Максимальное количество постоянных соединений на хост агента задаётся опцией persistent_connections_limit в секции searchd.

Важно отметить, что значение persistent_connections_limit должно быть больше 0 для использования постоянных соединений с агентом. Если оно не определено, по умолчанию равно 0, и директива agent_persistent будет работать так же, как agent.

Использование постоянных соединений мастер-агент снижает нагрузку на TCP-порты и экономит время на установку соединений, делая процесс более эффективным.

agent_blackhole = testbox:9312:testindex1,testindex2

Опции, связанные с удаленными агентами, поддерживаются не во всех областях видимости одновременно. На этой странице основное внимание уделяется семантике удаленных таблиц и отдельных агентов. Для поведения, специфичного для зеркал, см. Зеркалирование и Балансировка нагрузки. Для полного поведения на уровне демона используйте связанные справочные страницы в Настройках Searchd.

agent_connect_timeout = 300ms

CREATE TABLE products_dist type='distributed'
  agent='127.0.0.1:9312:products|127.0.0.1:9313:products'
  agent_connect_timeout='300ms';

agent_query_timeout = 10000 # our query can be long, allow up to 10 sec

SELECT * FROM products_dist OPTION agent_query_timeout=750;

agent_retry_count определяет, сколько раз Manticore попытается подключиться к удаленным агентам и выполнить запросы в распределенной таблице, прежде чем сообщить о фатальной ошибке запроса. Имя параметра зависит от области применения: используйте

• agent_retry_count как настройку уровня экземпляра,
• agent_retry_count или его псевдоним mirror_retry_count для распределенной таблицы,
• OPTION retry_count=... для каждого запроса,
• и [retry_count=...] внутри отдельного объявления agent=....

Полные детали на уровне демона приведены в разделе Searchd: agent_retry_count.

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

table products_dist {
  type = distributed
  agent = 127.0.0.1:9312|127.0.0.1:9313:products[retry_count=2]
}

SELECT * FROM products_dist OPTION retry_count=1;

agent_retry_delay = 500ms

SELECT * FROM products_dist OPTION retry_count=2, retry_delay=300;

client_timeout = 1h

Опция hostname_lookup определяет стратегию обновления имён хостов. По умолчанию IP-адреса host-имён агентов кешируются при запуске сервера, чтобы избежать чрезмерного обращения к DNS. Однако в некоторых случаях IP может динамически меняться (например, облачный хостинг), и может быть необходимо не кешировать IP. Установка этой опции в request отключает кеширование и выполняет запросы к DNS при каждом запросе. IP-адреса также можно обновлять вручную с помощью команды FLUSH HOSTNAMES.

Опция listen_tfo позволяет использовать флаг TCP_FASTOPEN для всех слушателей. По умолчанию она управляется системой, но может быть явно отключена, установив значение '0'.

Для получения дополнительной информации о расширении TCP Fast Open, пожалуйста, обратитесь к Wikipedia. Кратко, оно позволяет исключить один круг обмена TCP при установлении соединения.

На практике использование TFO может оптимизировать сетевую эффективность клиент-агента, аналогично использованию agent_persistent, но без удержания активных соединений и без ограничений на максимальное количество соединений.

Большинство современных операционных систем поддерживают TFO. Linux (как одна из самых прогрессивных) поддерживает его с 2011 года, начиная с ядра 3.7 (для серверной стороны). Windows поддерживает его с некоторых сборок Windows 10. Другие системы, такие как FreeBSD и MacOS, также участвуют.

Для систем Linux сервер проверяет переменную /proc/sys/net/ipv4/tcp_fastopen и действует соответственно. Бит 0 управляет клиентской стороной, а бит 1 управляет слушателями. По умолчанию система имеет этот параметр, установленный в 1, то есть клиенты включены, а слушатели отключены.

persistent_connections_limit = 29 # assume that each host of agents has max_connections = 30 (or 29).

Опция persistent_connections_limit определяет максимальное количество одновременных постоянных соединений с удалёнными постоянными агентами. Это настройка на уровне экземпляра и должна быть определена в секции конфигурации searchd. Каждый раз при установлении соединения с агентом, определённым в agent_persistent, происходит попытка повторно использовать существующее соединение (если оно есть) или создать новое и сохранить его для будущего использования. Однако в некоторых случаях может потребоваться ограничить количество постоянных соединений. Эта директива задаёт лимит и влияет на количество соединений с каждым хостом агента для всех распределённых таблиц.

Рекомендуется установить это значение равным или меньшим, чем опция max_connections в конфигурации агента.

Особый случай распределённой таблицы — одна локальная и несколько удалённых, которая используется исключительно для создания распределённых сниппетов, когда сниппеты берутся из файлов. В этом случае локальная таблица может выступать в роли "шаблонной" таблицы, обеспечивая настройки для токенизации при построении сниппетов.

snippets_file_prefix = /mnt/common/server1/

Опция snippets_file_prefix — это необязательный префикс, который можно добавить к именам локальных файлов при генерации сниппетов. Значение по умолчанию — текущая рабочая папка.

Чтобы узнать больше о создании распределённых сниппетов, смотрите CALL SNIPPETS.

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

Для DPQ операции перечисления сохранённых запросов и поиска по ним (с использованием CALL PQ) прозрачны и работают так, как если бы все таблицы были одной локальной таблицей. Однако операции манипуляции данными, такие как insert, replace, truncate, отсутствуют.

Если в список агентов включить не перколяционную таблицу, поведение будет неопределённым. Если неправильный агент имеет ту же схему, что и внешняя схема PQ таблицы (id, query, tags, filters), это не вызовет ошибку при перечислении сохранённых правил PQ и может засорить список фактических правил PQ, хранящихся в PQ таблицах, собственными не-PQ строками. В результате будьте осторожны и понимаете путаницу, которую это может вызвать. Вызов CALL PQ к такому неправильному агенту вызовет ошибку.

Для получения дополнительной информации о выполнении запросов к распределённой перколяционной таблице смотрите выполнение запросов к распределённой перколяционной таблице.

ПРИМЕЧАНИЕ: Автошардирование требует Manticore Buddy. Если автошардирование не работает, убедитесь, что Buddy установлен и запущен.

Manticore позволяет создавать шардированные таблицы — особый тип таблиц (type='shard'), который прозрачно распределяет чтение и запись между несколькими физическими шардами. Эта возможность особенно полезна при масштабировании данных. Можно создавать как локальные шардированные таблицы на одном сервере, так и реплицированные шардированные таблицы в многонодовом репликационном кластере, чтобы оптимизировать распределение данных.

Шардированные таблицы дают несколько ключевых преимуществ для высокопроизводительных приложений:

• Они обеспечивают более высокую производительность записи, потому что данные пишутся сразу в несколько шардов параллельно, а системные ресурсы используются эффективнее. Такая параллельная запись дает более высокую скорость индексации по сравнению с одной большой таблицей.
• Шардированные таблицы поддерживают репликацию из коробки, повышая доступность системы. Вам не нужно настраивать репликацию вручную. Просто задайте коэффициент репликации при создании шардированной таблицы, и система сама все сделает. Такая встроенная репликация обеспечивает непрерывный доступ к данным даже при отказе части узлов.

• shards='N' — число физических шардов, которые нужно создать. Обязательный параметр. Должно быть положительным целым числом, максимум 3000.
• rf='M' — коэффициент репликации: число копий, которое хранится для каждого шарда. Обязательный параметр. На одном узле он должен быть 1; в репликационном кластере — от 1 до числа узлов в кластере.
• timeout='S' — сколько времени в секундах ждать подготовки шардов при создании. По умолчанию 30. Увеличьте значение, если создаете много шардов на большом числе узлов.

Значения нужно указывать в кавычках (shards='10', а не shards=10). Имена параметров не чувствительны к регистру.

Чтобы создать локальную шардированную таблицу, создайте таблицу как обычно и добавьте shards='N' и rf='1'. shards — это число физических шардов, которые будут созданы за таблицей. rf — коэффициент репликации. На одном сервере он должен быть 1.

Вот пример, который создает таблицу с 10 шардами и автоматически распределяет данные между ними:

CREATE TABLE local_sharded shards='10' rf='1'

После этого запроса вы получаете одну шардированную таблицу, у которой все шарды уже настроены. Физические шарды находятся в базе system и скрыты из SHOW TABLES; вы работаете с шардированной таблицей по ее публичному имени, а система автоматически направляет операции к нужным шардам.

Чтобы защититься от отказов сервера, настройте репликационный кластер на тех узлах, которые должны участвовать. Далее в этой документации мы будем считать, что созданный вами кластер называется c. Добавьте все нужные узлы, следуя инструкциям по репликации, а затем создайте таблицу с нужным коэффициентом репликации.

Например, допустим, у вас есть 3-узловой репликационный кластер и вы хотите таблицу, разбитую на 10 шардов, с одной копией каждого шарда на каждом узле. При участии трех узлов нужно задать rf='3':

CREATE TABLE c:cluster_sharded shards='10' rf='3'

После этого можно работать с таблицей по ее обычному имени на любом узле кластера — для INSERT, SELECT, UPDATE и DELETE не нужен префикс кластера. Префикс кластера (c:) используется только для DDL, например CREATE TABLE и DROP TABLE.

Стандартное время ожидания завершения всех процессов подготовки шардов при создании — 30 секунд. Иногда при создании большого числа шардов в репликационном кластере с несколькими узлами это занимает чуть больше времени из-за сетевой задержки. При необходимости это время можно увеличить с помощью параметра timeout:

CREATE TABLE c:cluster_sharded shards='10' rf='3' timeout='60'

Если время ожидания будет превышено, создание таблицы завершится с ошибкой, и вам придется повторить попытку с большим значением timeout.

Чтобы удалить шардированную таблицу, используйте стандартную команду DROP TABLE. В кластерной среде указывайте имя кластера в имени таблицы, чтобы точно обратиться к нужной таблице. Поддерживается IF EXISTS.

Чтобы удалить локальную шардированную таблицу:

DROP TABLE local_sharded
DROP TABLE IF EXISTS local_sharded

Чтобы удалить реплицированную шардированную таблицу:

DROP TABLE c:cluster_sharded
DROP TABLE IF EXISTS c:cluster_sharded

DESC <table> возвращает пользовательскую схему полей шардированной таблицы (те столбцы, которые вы объявили).

SHOW CREATE TABLE <table> возвращает пользовательское определение с shards='N' rf='M' — внутренняя топология type='shard' (клаузулы local=/agent= для каждого шарда и репликационный кластер с md5-именем) намеренно скрыта. Чтобы посмотреть раскрытую внутреннюю топологию для диагностики, используйте SHOW CREATE TABLE <table> OPTION force=1.

Для проверки состояния подсистемы шардирования предусмотрены две дополнительные команды:

• SHOW SHARDING STATUS [[<cluster>:]<table>] — показывает размещение и состояние каждого шарда. Без аргументов выводит все шардированные таблицы; с именем таблицы (при желании с префиксом кластера) фильтрует вывод только по ней. Возвращаемые столбцы: table, shard, node, status (active/inactive), cluster, replication_cluster, rf и rf_status (ok/degraded/broken).

   SHOW SHARDING STATUS
   SHOW SHARDING STATUS cluster_sharded
   SHOW SHARDING STATUS c:cluster_sharded

• SHOW SHARDING MASTER — показывает, на каком узле сейчас запущен процесс sharding master, и находится ли он в состоянии active или inactive.

   SHOW SHARDING MASTER