A remote table in Manticore Search is represented by the agent prefix in the definition of a distributed table. A distributed table can include a combination of local and remote tables. If there are no local tables provided, the distributed table will be purely remote and serve as a proxy only. For example, you might have a Manticore instance that listens on multiple ports and serves different protocols, and then redirects queries to backend servers that only accept connections via Manticore's internal binary protocol, using persistent connections to reduce the overhead of establishing connections. Even though a purely remote distributed table doesn't serve local tables itself, it still consumes machine resources, as it still needs to perform final calculations, such as merging results and calculating final aggregated values.

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

agent directive declares the remote agents that are searched each time the enclosing distributed table is searched. These agents are essentially pointers to networked tables. The value specified includes the address and can also include multiple alternatives (agent mirrors) for either the address only or the address and table list. See Mirroring for mirror syntax and Load balancing for how mirrored agents are selected.

The address specification must be one of the following:

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

The hostname is the remote host name, port is the remote TCP port number, table-list is a comma-separated list of table names, and square brackets [] indicate an optional clause. For TCP connections this port is the remote agent/API port (typically 9312), not the MySQL port (9306).

If the table name is omitted, it is assumed to be the same table as the one where this line is defined. In other words, when defining agents for the 'mycoolindex' distributed table, you can simply point to the address, and it will be assumed that you are querying the mycoolindex table on the agent's endpoints.

If the port number is omitted, it is assumed to be 9312. If it is defined but invalid (e.g. 70000), the agent will be skipped.

You can point each agent to one or more remote tables residing on one or more networked servers with no restrictions. This enables several different usage modes:

• Sharding over multiple agent servers and creating an arbitrary cluster topology
• Sharding over multiple agent servers mirrored for high availability and load balancing purposes (see Mirroring and Load balancing)
• Sharding within localhost to utilize multiple cores (however, it is simpler just to use multiple local tables)

To avoid confusion:

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

All agents are searched in parallel. The index list is passed verbatim to the remote agent. The exact way that list is searched within the agent (i.e. sequentially or in parallel) depends solely on the agent's configuration (see the threads setting). The master has no remote control over this.

It is important to note that the LIMIT, option is ignored in agent queries. This is because each agent can contain different tables, so it is the responsibility of the client to apply the limit to the final result set. This is why the query to a physical table is different from the query to a distributed table when viewed in the query logs. The query cannot be a simple copy of the original query, as this would not produce the correct results.

For example, if a client makes a query SELECT ... LIMIT 10, 10, and there are two agents, with the second agent having only 10 documents, broadcasting the original LIMIT 10, 10 query would result in receiving 0 documents from the second agent. However, LIMIT 10,10 should return documents 10-20 from the resulting set. To resolve this, the query must be sent to the agents with a broader limit, such as the default max_matches value of 1000.

For instance, if there is a distributed table dist that refers to the remote table user, a client query SELECT * FROM dist LIMIT 10,10 would be converted to SELECT * FROM user LIMIT 0,1000 and sent to the remote table user. Once the distributed table receives the result, it will apply the LIMIT 10,10 and return the requested 10 documents.

SELECT * FROM dist LIMIT 10,10;

the query will be converted to:

SELECT * FROM user LIMIT 0,1000

Additionally, the value can specify options for each individual agent, such as:

• ha_strategy - random, roundrobin, nodeads, noerrors (overrides the global ha_strategy setting for the particular agent; see also Mirroring)
• conn - pconn, persistent (equivalent to setting agent_persistent at the table level)
• blackhole 0,1 (identical to the agent_blackhole setting for the agent)
• retry_count an integer value (corresponding to agent_retry_count , but the provided value will not be multiplied by the number of mirrors)

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

Example:

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

For optimal performance, it's recommended to place remote tables that reside on the same server within the same record. For instance, instead of:

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

you should prefer:

agent = remote:9312:idx1,idx2

agent_persistent = remotebox:9312:index2

The agent_persistent option allows you to persistently connect to an agent, meaning the connection will not be dropped after a query is executed. The syntax for this directive is the same as the agent directive. However, instead of opening a new connection to the agent for each query and then closing it, the master will keep a connection open and reuse it for subsequent queries. The maximum number of persistent connections per agent host is defined by the persistent_connections_limit option in the searchd section.

It's important to note that the persistent_connections_limit must be set to a value greater than 0 in order to use persistent agent connections. If it's not defined, it defaults to 0, and the agent_persistent directive will act the same as the agentdirective.

Using persistent master-agent connections reduces TCP port pressure and saves time on connection handshakes, making it more efficient.

agent_blackhole = testbox:9312:testindex1,testindex2

The options related to remote agents are not all supported at the same scope. This page focuses on remote-table and per-agent semantics. For mirror-specific behavior, see Mirroring and Load balancing. For full daemon-level behavior, use the linked reference pages in Searchd settings.

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 specifies how many times Manticore will attempt to connect to and query remote agents in a distributed table before reporting a fatal query error. The name varies by scope: use

• agent_retry_count as the instance-wide setting,
• agent_retry_count or its alias mirror_retry_count on a distributed table,
• OPTION retry_count=... per query,
• and [retry_count=...] inside an individual agent=... declaration.

Full daemon-level details are in Searchd: agent_retry_count.

If you use agent mirrors, the server selects a different mirror before each connection attempt according to ha_strategy, and agent_retry_count is aggregated across mirrors.

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

The hostname_lookup option defines the strategy for renewing hostnames. By default, the IP addresses of agent host names are cached at server start to avoid excessive access to DNS. However, in some cases, the IP can change dynamically (e.g. cloud hosting) and it may be desirable to not cache the IPs. Setting this option to request disables the caching and queries the DNS for each query. The IP addresses can also be manually renewed using the FLUSH HOSTNAMES command.

The listen_tfo option allows for the use of the TCP_FASTOPEN flag for all listeners. By default, it is managed by the system, but it can be explicitly turned off by setting it to '0'.

For more information about the TCP Fast Open extension, please refer to Wikipedia. In short, it allows to eliminate one TCP round-trip when establishing a connection.

In practice, using TFO can optimize the client-agent network efficiency, similar to when agent_persistent is in use, but without holding active connections and without limitations on the maximum number of connections.

Most modern operating systems support TFO. Linux (as one of the most progressive) has supported it since 2011, with kernels starting from 3.7 (for the server side). Windows has supported it since some builds of Windows 10. Other systems, such as FreeBSD and MacOS, are also in the game.

For Linux systems, the server checks the variable /proc/sys/net/ipv4/tcp_fastopen and behaves accordingly. Bit 0 manages the client side, while bit 1 rules the listeners. By default, the system has this parameter set to 1, i.e., clients are enabled and listeners are disabled.

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

The persistent_connections_limit option defines the maximum number of simultaneous persistent connections to remote persistent agents. This is an instance-wide setting and must be defined in the searchd configuration section. Each time a connection to an agent defined under agent_persistent is made, we attempt to reuse an existing connection (if one exists) or create a new connection and save it for future use. However, in some cases it may be necessary to limit the number of persistent connections. This directive defines the limit and affects the number of connections to each agent's host across all distributed tables.

It is recommended to set this value equal to or less than the max_connections option in the agent's configuration.

A special case of a distributed table is a single local and multiple remotes, which is used exclusively for distributed snippets creation, when snippets are sourced from files. In this case, the local table may act as a "template" table, providing settings for tokenization when building snippets.

snippets_file_prefix = /mnt/common/server1/

The snippets_file_prefix is an optional prefix that can be added to the local file names when generating snippets. The default value is the current working folder.

To learn more about distributed snippets creation, see CALL SNIPPETS.

You can create a distributed table from multiple percolate tables. The syntax for constructing this type of table is the same as for other distributed tables, and can include multiplelocal tables as well as agents.

For DPQ, the operations of listing stored queries and searching through them (using CALL PQ) are transparent and work as if all the tables were one single local table. However, data manipulation statements such as insert, replace, truncate are not available.

If you include a non-percolate table in the list of agents, the behavior will be undefined. If the incorrect agent has the same schema as the outer schema of the PQ table (id, query, tags, filters), it will not trigger an error when listing stored PQ rules, and may pollute the list of actual PQ rules stored in PQ tables with its own non-PQ strings. As a result, be cautious and aware of the confusion that this may cause. ACALL PQ to such an incorrect agent will trigger an error.

For more information on making queries to a distributed percolate table, see making queries to a distribute percolate table.

NOTE: Auto-sharding requires Manticore Buddy. If auto-sharding doesn't work, make sure Buddy is installed and running.

Manticore allows for the creation of sharded tables, a special table type (type='shard') that transparently fans out reads and writes across multiple underlying physical shards. This feature proves invaluable when scaling your data. You can create both local sharded tables on a single server and replicated sharded tables on a multi-node replication cluster to optimize data distribution.

Sharded tables offer several key benefits for high-performance applications:

• They deliver superior write performance by writing data to multiple shards in parallel, utilizing system resources more efficiently. This parallel writing capability enables higher indexing throughput compared to a single large table.
• Sharded tables support replication out of the box, enhancing high availability. You don't need to handle replication manually. Simply set the replication factor when creating the sharded table, and the system manages everything. This built-in replication ensures continuous data accessibility even if some nodes fail.

• shards='N' — number of physical shards to create. Required. Must be a positive integer, with a maximum of 3000.
• rf='M' — replication factor: the number of copies kept for each shard. Required. On a single node it must be 1; on a replication cluster it must be between 1 and the number of nodes in the cluster.
• timeout='S' — how long (in seconds) to wait for shard preparation during creation. Defaults to 30. Increase it when creating many shards across many nodes.

Values must be quoted (shards='10', not shards=10). Option names are case-insensitive.

To create a local sharded table, create a table as you normally would and add shards='N' and rf='1'. shards is the number of physical shards that will be created behind the table. rf is the replication factor. On a single server it must be 1.

Here's an example that creates a table with 10 shards, with data automatically distributed across them:

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

After this query you get a single sharded table with all its shards already set up. The underlying physical shards live under the system database and are hidden from SHOW TABLES; you interact with the sharded table by its public name and the system routes operations to the appropriate shards automatically.

To protect against server outages, set up a replication cluster across the nodes you want to participate. Throughout this documentation we'll assume the cluster you created is named c. Add all desired nodes by following the replication instructions, then create the table with the desired replication factor.

For example, let's assume you have a 3-node replication cluster and want a table sharded into 10 shards with one copy of each shard on every node. With three nodes participating you set rf='3':

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

After that you can work with the table by its plain name on any cluster node — INSERT, SELECT, UPDATE, and DELETE do not require the cluster prefix. The cluster prefix (c:) is only used for DDL such as CREATE TABLE and DROP TABLE.

The default timeout to wait for all processes of shard preparation during creation is 30 seconds. Sometimes, when creating many shards on a replication cluster with multiple nodes, it takes a bit longer due to network latency. If needed, you can increase this timeout via the timeout option:

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

If the timeout is exceeded, the table creation will fail, and you'll need to retry with a longer timeout value.

To drop a sharded table, use the standard DROP TABLE command. In a clustered environment, specify the cluster name in the table name to properly target the table you want to drop. IF EXISTS is supported.

To delete a local sharded table:

DROP TABLE local_sharded
DROP TABLE IF EXISTS local_sharded

To delete a replicated sharded table:

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

DESC <table> returns the user-facing field schema of a sharded table (the columns you declared).

SHOW CREATE TABLE <table> returns the user-facing definition with shards='N' rf='M' — the internal type='shard' topology (the per-shard local=/agent= clauses and md5-named replication cluster) is intentionally hidden. To inspect the resolved internal topology for diagnostics, use SHOW CREATE TABLE <table> OPTION force=1.

Two extra commands are provided to inspect the state of the sharding subsystem:

• SHOW SHARDING STATUS [[<cluster>:]<table>] — lists per-shard placement and health. Without arguments it lists all sharded tables; with a table name (optionally cluster-prefixed) it filters to that table. Returned columns: table, shard, node, status (active/inactive), cluster, replication_cluster, rf, and rf_status (ok/degraded/broken).

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

• SHOW SHARDING MASTER — shows which node currently runs the sharding master process and whether it's active or inactive.

   SHOW SHARDING MASTER

When working with sharded tables, be aware of the following limitations:

1. Local and Clustered Sharded Tables Cannot Coexist on the Same Nodes: On a set of nodes that participate in a replication cluster, you cannot mix:

   • Local sharded tables (created without a cluster prefix: create table s ... shards='N' rf='1')
   • Clustered sharded tables (created with a cluster prefix: create table c:r ... shards='N' rf='M')

   Example: Consider a 2-node cluster where:

   • Table s was created independently on each node as a local sharded table
   • Table r is replicated across both nodes via the cluster

   In this setup, attempting to create a clustered sharded table (create table c:r ... shards='N' rf='2') on the same nodes will fail.

2. Cluster Name Consistency:

   • The sharding subsystem binds to a single cluster name on first use; all clustered sharded tables on these nodes must use that same cluster name.
   • Once the cluster name is chosen, it applies to all subsequent clustered sharded table creations.

   Example: If you create your first clustered sharded table with:

   create table c:users ... shards='N' rf='M'

   All subsequent clustered sharded tables must reuse cluster c:

   create table c:orders ... shards='N' rf='M'  -- works
   create table d:items  ... shards='N' rf='M'  -- fails

3. No Table Alterations:

   • Once a sharded table is created, its structure cannot be modified with ALTER TABLE.
   • To change the schema, you must:
     1. Create a new sharded table with the desired structure
     2. Copy the data to the new table
     3. Drop the old table

4. Shard Count Limit:

   • Maximum of 3,000 shards per sharded table.
   • This limit applies regardless of cluster configuration or table size.
   • Plan your sharding strategy accordingly to stay within this limit.

5. rf and shards Must Be Quoted Integers:

   • Both options require a quoted numeric value (shards='10', rf='2'); unquoted, non-numeric, empty, or fractional values are rejected.
   • On a standalone (non-clustered) server, rf must be '1'.
   • On a replication cluster, rf must be between 1 and the number of nodes in the cluster.