To understand how to add a distributed table with remote agents, it is important to first have a basic understanding of distributed tables In this article, we will focus on how to use a distributed table as the basis for creating a cluster of Manticore instances.

table mydist {
          type  = distributed
          agent = box1:9312:shard1
          agent = box2:9312:shard2
          agent = box3:9312:shard3
          agent = box4:9312:shard4
}

In the event of a server failure, the distributed table will still work, but the results from the failed shard will be missing.

table mydist {
          type  = distributed
          agent = box1:9312|box5:9312:shard1
          agent = box2:9312:|box6:9312:shard2
          agent = box3:9312:|box7:9312:shard3
          agent = box4:9312:|box8:9312:shard4
}

Load balancing is turned on by default for any distributed table that uses mirroring. By default, queries are distributed randomly among mirrors. You can change this behavior by using ha_strategy.

ha_strategy = {random|nodeads|noerrors|roundrobin}

The mirror selection strategy for load balancing is optional and is set to random by default.

ha_strategy can be set globally in the searchd configuration or per distributed table. The strategy used for mirror selection, or in other words, choosing a specific agent mirror in a distributed table, is controlled by this directive. In practice, this directive controls how the master balances requests between mirrored agent nodes. The following strategies are implemented:

ha_strategy = random

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

The default simple random strategy does not take into account the status of mirrors, error rates, and, most importantly, actual response latencies. To address heterogeneous clusters and temporary spikes in agent node load, there is a group of balancing strategies that dynamically adjust the probabilities based on the query latencies observed by the master.

The adaptive strategies based on latency-weighted probabilities work as follows. The size of that statistics window is controlled by ha_period_karma, while idle mirror health and round-trip tracking are driven by ha_ping_interval. Both are described later on this page and in Searchd settings.

1. Latency stats are accumulated in blocks of ha_period_karma seconds.
2. Latency-weighted probabilities are recomputed once per karma period.
3. The "dead or alive" flag is adjusted once per request, including ping requests.

Initially, the probabilities are equal. On every step, they are scaled by the inverse of the latencies observed during the last karma period, and then renormalized. For example, if during the first 60 seconds after the master startup, 4 mirrors had latencies of 10 ms, 5 ms, 30 ms, and 3 ms respectively, the first adjustment step would go as follows:

1. Initial percentages: 0.25, 0.25, 0.25, 0.25.
2. Observed latencies: 10 ms, 5 ms, 30 ms, 3 ms.
3. Inverse latencies: 0.1, 0.2, 0.0333, 0.333.
4. Scaled percentages: 0.025, 0.05, 0.008333, 0.0833.
5. Renormalized percentages: 0.15, 0.30, 0.05, 0.50.

This means that the first mirror would have a 15% chance of being chosen during the next karma period, the second one a 30% chance, the third one (slowest at 30 ms) only a 5% chance, and the fourth and fastest one (at 3 ms) a 50% chance. After that period, the second adjustment step would update those chances again, and so on.

The idea is that once the observed latencies stabilize, the latency-weighted probabilities will stabilize as well. All these adjustment iterations are meant to converge at a point where the average latencies are roughly equal across all mirrors.

ha_strategy = nodeads

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

ha_strategy = noerrors

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

ha_strategy = roundrobin

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

Assume the mirrored distributed tables already exist, for example:

CREATE TABLE products_random type='distributed'
  agent='127.0.0.1:9312:products|127.0.0.1:9313:products';
CREATE TABLE products_rr type='distributed'
  agent='127.0.0.1:9312:products|127.0.0.1:9313:products'
  ha_strategy='roundrobin';

SELECT node FROM products_random;
SELECT node FROM products_random;
SELECT node FROM products_rr;
SELECT node FROM products_rr;

+-------+
| node  |
+-------+
| node2 |
+-------+
+-------+
| node  |
+-------+
| node1 |
+-------+
+-------+
| node  |
+-------+
| node1 |
+-------+
+-------+
| node  |
+-------+
| node2 |
+-------+

SHOW CREATE TABLE products_rr;
SHOW AGENT STATUS;
SHOW AGENT STATUS LIKE '%errorsarow%';
SHOW AGENT STATUS LIKE '%1periods_%';

+---------------------------------+-------+
| Key                             | Value |
+---------------------------------+-------+
| ag_0_pingtripmsec               | 0.233 |
| ag_0_errorsarow                 | 0     |
| ag_0_1periods_network_errors    | 0     |
| ag_0_1periods_warnings          | 0     |
| ag_0_1periods_succeeded_queries | 11    |
| ag_0_1periods_msecsperquery     | n/a   |
+---------------------------------+-------+

Practical interpretation:

• If errorsarow climbs above 3, nodeads starts treating that mirror as dead.
• If a mirror has no recent succeeded_queries, noerrors will skip it entirely.
• If several mirrors have low error ratios, noerrors still prefers the ones with lower latency because the normal weight rebalancing stays in play.
• pingtripmsec helps you distinguish "reachable but slow" from "failing" mirrors.

The options related to mirroring, timeouts, and retries are not all supported at the same scope. This page focuses on how they affect load balancing. For full syntax and daemon-level behavior, use the linked reference pages.

ha_strategy = random
table products_rr {
  type = distributed
  agent = 127.0.0.1:9312|127.0.0.1:9313:products
  ha_strategy = roundrobin
}

CREATE TABLE products_rr type='distributed'
  agent='127.0.0.1:9312:products|127.0.0.1:9313:products'
  ha_strategy='roundrobin';

ha_period_karma = 2m

ha_ping_interval = 3s

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 = 500ms

SELECT * FROM products_dist OPTION agent_query_timeout=750;

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;

Agent mirrors can be used interchangeably when processing a search query. The Manticore instance hosting the distributed table where the mirrored agents are defined keeps track of mirror status (alive or dead) and response times, and performs automatic failover and load balancing based on this information.

agent = node1|node2|node3:9312:shard2

The above example declares that node1:9312, node2:9312, and node3:9312 all have a table called shard2, and can be used as interchangeable mirrors. If any of these servers go down, the queries will be distributed between the remaining ones. When the server comes back online, the master will detect it and begin routing queries to all mirrors again.

A mirror may also include an individual table list, as follows:

agent = node1:9312:node1shard2|node2:9312:node2shard2

This works similarly to the previous example, but different table names will be used when querying different servers. For example, node1shard2 will be used when querying node1:9312, and node2shard2 will be used when querying node2:9312.

By default, mirror selection uses the global or table-level ha_strategy. If you do not set it explicitly, the default strategy is random. The master stores metrics such as total query count, error count, and response time for each agent and groups them into time spans controlled by ha_period_karma. These statistics are then used by the balancing strategies described in Load balancing.

The karma period is in seconds and defaults to 60 seconds. The master stores up to 15 karma spans with per-agent statistics for instrumentation purposes (see SHOW AGENT STATUS). However, only the last two spans are used for HA/LB logic.

When there are no queries, the master sends a regular ping command every ha_ping_interval in order to collect statistics and check if the remote host is still alive. ha_ping_interval defaults to 1000 msec. Setting it to 0 disables pings, and statistics will only be accumulated based on actual queries. Together with ha_period_karma, this controls how quickly mirror health and latency changes influence load balancing.

This distinction is important:

• One agent='host1|host2:table' entry means one remote shard with mirrored backends.
• Multiple agent='...' entries mean multiple remote shards.

For example:

# one shard, two mirrors
agent = node1|node2:9312:products
# two shards
agent = node1:9312:products_a
agent = node2:9312:products_b

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

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

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

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

+------+------------+-------+
| 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|... |
+---------------+----------------------------------------------------------------------------------+

# node1, node2 carry shard1 as local
# node3, node4 carry shard2 as local
# config on node1, node2
agent = node3:9312|node4:9312:shard2
# config on node3, node4
agent = node1:9312|node2:9312:shard1

With Manticore, write transactions (such as INSERT, REPLACE, DELETE, TRUNCATE, UPDATE, COMMIT) can be replicated to other cluster nodes before the transaction is fully applied on the current node. Currently, replication is supported for percolate, rt and distributed tables in Linux and macOS.

Native Windows binaries for Manticore do not support replication. We recommend installing Manticore via WSL (Windows Subsystem for Linux).

On macOS, replication has limited support and is recommended only for development purposes.

Manticore's replication is powered by the Galera library and boasts several impressive features:

• True multi-master: read and write to any node at any time.
• Virtually synchronous replication no slave lag and no data loss after a node crash.
• Hot standby: no downtime during failover (since there is no failover).
• Tightly coupled: all nodes hold the same state and no diverged data between nodes is allowed.
• Automatic node provisioning: no need to manually backup the database and restore it on a new node.
• Easy to use and deploy.
• Detection and automatic eviction of unreliable nodes.
• Certification-based replication.

To set up replication in Manticore Search:

• The data_dir option must be set in the "searchd" section of the configuration file. Replication is not supported in plain mode.
• A listen directive must be specified, containing an IP address accessible by other nodes, or a node_address with an accessible IP address.
• Optionally, you can set unique values for server_id on each cluster node. If no value is set, the node will attempt to use the MAC address or a random number to generate the server_id.

If there is no replication listen directive set, Manticore will use the first two free ports in the range of 200 ports after the default protocol listening port for each created cluster. To set replication ports manually, the listen directive (of replication type) port range must be defined and the address/port range pairs must not intersect between different nodes on the same server. As a rule of thumb, the port range should specify at least two ports per cluster. When you define a replication listener with a port range (e.g., listen = 192.168.0.1:9320-9328:replication), Manticore doesn't immediately start listening on these ports. Instead, it will take random free ports from the specified range only when you start using replication.

A replication cluster is a group of nodes in which a write transaction is replicated. Replication is set up on a per-table basis, meaning that one table can only belong to one cluster. There is no limit on the number of tables that a cluster can have. All transactions such as INSERT, REPLACE, DELETE, TRUNCATE on any percolate or real-time table that belongs to a cluster are replicated to all the other nodes in that cluster. Distributed tables can also be part of the replication process. Replication is multi-master, so writes to any node or multiple nodes simultaneously will work just as well.

To create a cluster, you can typically use the command create cluster with CREATE CLUSTER <cluster name>, and to join a cluster, you can use join cluster with JOIN CLUSTER <cluster name> at 'host:port'. However, in some rare cases, you may want to fine-tune the behavior of CREATE/JOIN CLUSTER. The available options are:

This option specifies the name of the cluster. It should be unique among all the clusters in the system.

Note: The maximum allowable hostname length for the JOIN command is 253 characters. If you exceed this limit, searchd will generate an error.

The path option specifies the data directory for write-set cache replication and incoming tables from other nodes. This value should be unique among all the clusters in the system and should be specified as a relative path to the data_dir. directory. By default, it is set to the value of data_dir.

The nodes option is a list of address:port pairs for all the nodes in the cluster, separated by commas. This list should be obtained using the node's API interface and can include the address of the current node as well. It is used to join the node to the cluster and to rejoin it after a restart.

The options option allows you to pass additional options directly to the Galera replication plugin, as described in the Galera Documentation Parameters

INSERT INTO posts:weekly_index VALUES ( 'iphone case' )
TRUNCATE RTINDEX click_query:weekly_index
UPDATE INTO posts:rt_tags SET tags=(101, 302, 304) WHERE MATCH ('use') AND id IN (1,101,201)
DELETE FROM clicks:rt WHERE MATCH ('dumy') AND gid>206

POST /insert -d '
{
  "cluster":"posts",
  "table":"weekly_index",
  "doc":
  {
    "title" : "iphone case",
    "price" : 19.85
  }
}'
POST /delete -d '
{
  "cluster":"posts",
  "table": "weekly_index",
  "id":1
}'

$index->addDocuments([
        1, ['title' => 'iphone case', 'price' => 19.85]
]);
$index->deleteDocument(1);

indexApi.insert({"cluster":"posts","table":"weekly_index","doc":{"title":"iphone case","price":19.85}})
indexApi.delete({"cluster":"posts","table":"weekly_index","id":1})

await indexApi.insert({"cluster":"posts","table":"weekly_index","doc":{"title":"iphone case","price":19.85}})
await indexApi.delete({"cluster":"posts","table":"weekly_index","id":1})

res = await indexApi.insert({"cluster":"posts","table":"weekly_index","doc":{"title":"iphone case","price":19.85}});
 res = await indexApi.delete({"cluster":"posts","table":"weekly_index","id":1});

InsertDocumentRequest newdoc = new InsertDocumentRequest();
HashMap<String,Object> doc = new HashMap<String,Object>(){{
    put("title","Crossbody Bag with Tassel");
    put("price",19.85);
}};
newdoc.index("weekly_index").cluster("posts").id(1L).setDoc(doc);
sqlresult = indexApi.insert(newdoc);
DeleteDocumentRequest deleteRequest = new DeleteDocumentRequest();
deleteRequest.index("weekly_index").cluster("posts").setId(1L);
indexApi.delete(deleteRequest);

Dictionary<string, Object> doc = new Dictionary<string, Object>();
doc.Add("title", "Crossbody Bag with Tassel");
doc.Add("price", 19.85);
InsertDocumentRequest newdoc = new InsertDocumentRequest(table: "weekly_index", cluster:posts, id: 1, doc: doc);
var sqlresult = indexApi.Insert(newdoc);
DeleteDocumentRequest deleteDocumentRequest = new DeleteDocumentRequest(table: "weekly_index", cluster: "posts", id: 1);
indexApi.Delete(deleteDocumentRequest);

let mut doc = HashMap::new();
doc.insert("title".to_string(), serde_json::json!("Crossbody Bag with Tassel"));
doc.insert("price".to_string(), serde_json::json!(19.85));
let insert_req = InsertDocumentRequest {
    table: serde_json::json!("weekly_index"),
    doc: serde_json::json!(doc),
    cluster: serde_json::json!("posts"),
    id: serde_json::json!(1),
};
let insert_res = index_api.insert(insert_req).await;
let delete_req = DeleteDocumentRequest {
    table: serde_json::json!("weekly_index"),
    cluster: serde_json::json!("posts"),
    id: serde_json::json!(1),
};
index_api.delete(delete_req).await;

SELECT * FROM weekly_index
CALL PQ('posts:weekly_index', 'document is here')

POST /search -d '
{
  "cluster":"posts",
  "table":"weekly_index",
  "query":{"match":{"title":"keyword"}}
}'
POST /search -d '
{
  "table":"weekly_index",
  "query":{"match":{"title":"keyword"}}
}'

SET CLUSTER click_query GLOBAL 'pc.bootstrap' = 1

POST /cli -d "
SET CLUSTER click_query GLOBAL 'pc.bootstrap' = 1
"

SET CLUSTER posts GLOBAL 'pc.bootstrap' = 1

POST /cli -d "
SET CLUSTER posts GLOBAL 'pc.bootstrap' = 1
"

searchd {
  listen   = 9312
  listen   = 192.168.1.101:9360-9370:replication
  data_dir = /var/lib/manticore/
  ...
 }

CREATE CLUSTER posts

POST /cli -d "
CREATE CLUSTER posts
"

$params = [
    'cluster' => 'posts'
    ]
];
$response = $client->cluster()->create($params);

utilsApi.sql('CREATE CLUSTER posts')

await utilsApi.sql('CREATE CLUSTER posts')

res = await utilsApi.sql('CREATE CLUSTER posts');

utilsApi.sql("CREATE CLUSTER posts");

utilsApi.Sql("CREATE CLUSTER posts");

utils_api.sql("CREATE CLUSTER posts", Some(true)).await;

ALTER CLUSTER posts ADD pq_title
ALTER CLUSTER posts ADD pq_clicks

POST /cli -d "
ALTER CLUSTER posts ADD pq_title
"
POST /cli -d "
ALTER CLUSTER posts ADD pq_clicks
"

$params = [
  'cluster' => 'posts',
  'body' => [
     'operation' => 'add',
     'table' => 'pq_title'
  ]
];
$response = $client->cluster()->alter($params);
$params = [
  'cluster' => 'posts',
  'body' => [
     'operation' => 'add',
     'table' => 'pq_clicks'
  ]
];
$response = $client->cluster()->alter($params);

utilsApi.sql('ALTER CLUSTER posts ADD pq_title')
utilsApi.sql('ALTER CLUSTER posts ADD pq_clicks')

await utilsApi.sql('ALTER CLUSTER posts ADD pq_title')
await utilsApi.sql('ALTER CLUSTER posts ADD pq_clicks')

res = await utilsApi.sql('ALTER CLUSTER posts ADD pq_title');
res = await utilsApi.sql('ALTER CLUSTER posts ADD pq_clicks');

utilsApi.sql("ALTER CLUSTER posts ADD pq_title");
utilsApi.sql("ALTER CLUSTER posts ADD pq_clicks");

utilsApi.Sql("ALTER CLUSTER posts ADD pq_title");
utilsApi.Sql("ALTER CLUSTER posts ADD pq_clicks");

utils_api.sql("ALTER CLUSTER posts ADD pq_title", Some(true)).await;
utils_api.sql("ALTER CLUSTER posts ADD pq_clicks", Some(true)).await;

JOIN CLUSTER posts AT '192.168.1.101:9312'

POST /cli -d "
JOIN CLUSTER posts AT '192.168.1.101:9312'
"

$params = [
  'cluster' => 'posts',
  'body' => [
      '192.168.1.101:9312'
  ]
];
$response = $client->cluster->join($params);

utilsApi.sql('JOIN CLUSTER posts AT \'192.168.1.101:9312\'')

await utilsApi.sql('JOIN CLUSTER posts AT \'192.168.1.101:9312\'')

res = await utilsApi.sql('JOIN CLUSTER posts AT \'192.168.1.101:9312\'');

utilsApi.sql("JOIN CLUSTER posts AT '192.168.1.101:9312'");

utilsApi.Sql("JOIN CLUSTER posts AT '192.168.1.101:9312'");

utils_api.sql("JOIN CLUSTER posts AT '192.168.1.101:9312'", Some(true)).await;

INSERT INTO posts:pq_title VALUES ( 3, 'test me' )

POST /insert -d '
{
  "cluster":"posts",
  "table":"pq_title",
  "id": 3
  "doc":
  {
    "title" : "test me"
  }
}'

$index->addDocuments([
        3, ['title' => 'test me']
]);

indexApi.insert({"cluster":"posts","table":"pq_title","id":3"doc":{"title":"test me"}})

await indexApi.insert({"cluster":"posts","table":"pq_title","id":3"doc":{"title":"test me"}})

res = await indexApi.insert({"cluster":"posts","table":"pq_title","id":3"doc":{"title":"test me"}});

InsertDocumentRequest newdoc = new InsertDocumentRequest();
HashMap<String,Object> doc = new HashMap<String,Object>(){{
    put("title","test me");
}};
newdoc.index("pq_title").cluster("posts").id(3L).setDoc(doc);
sqlresult = indexApi.insert(newdoc);

Dictionary<string, Object> doc = new Dictionary<string, Object>();
doc.Add("title", "test me");
InsertDocumentRequest newdoc = new InsertDocumentRequest(index: "pq_title", cluster: "posts", id: 3, doc: doc);
var sqlresult = indexApi.Insert(newdoc);

let mut doc = HashMap::new();
doc.insert("title".to_string(), serde_json::json!("test me"));
let insert_req = InsertDocumentRequest {
    table: serde_json::json!("pq_title"),
    doc: serde_json::json!(doc),
    cluster: serde_json::json!("posts"),
    id: serde_json::json!(3),
};
let insert_res = index_api.insert(insert_req).await;

All queries that modify tables in the cluster are now replicated to all nodes in the cluster.