Shard Splitting

• US 20200133550A1
• Filed: 10/31/2018
• Published: 04/30/2020
• Est. Priority Date: 10/31/2018
• Status: Active Grant

• Current Assignee: Elasticsearch Global B.V. (Elastic N.V.)
• Original Assignee: Elasticsearch Global B.V. (Elastic N.V.)
• Inventors: Willnauer, Simon Daniel
• Granted Patent: US 11,461,270 B2
• Time in Patent Office: Days
• CPC Class Codes: G06F 16/137 Hash-based content-based indexing of textual data G06F16/31
• CPC Class Codes: G06F 16/1827 Management specifically adapted to NAS management of storage area networks [SAN] G06F3/067
• CPC Class Codes: G06F 3/0619 in relation to data integrity, e.g. data losses, bit errors
• CPC Class Codes: G06F 3/065 Replication mechanisms
• CPC Class Codes: G06F 3/067 Distributed or networked storage systems, e.g. storage area networks [SAN], network attached storage [NAS]

Abstract

Methods and systems for are provided. Exemplary methods include: marking a source index as read only, the source index comprising a source shard, the source shard comprising a source reference; creating a target index, the target index comprising target shards, each target shard of the target shards comprising a target reference of target references; copying the source reference, the copying producing the target references; hashing identifiers in the source reference, each identifier being associated with a document of a plurality of documents of the source shard, the hashing assigning each document of the plurality of documents to a target shard of the target shards, the plurality of documents being stored in a file associated with the source reference; hard linking the file into the target references; marking the target index as read-write; and deleting the source index.

Claims

1. A computer-implemented method for splitting a shard comprising:

marking a source index as read only, the source index comprising a source shard, the source shard comprising a source reference;

creating a target index, the target index comprising target shards, each target shard of the target shards comprising a target reference of target references;

copying the source reference, the copying producing the target references;

hashing identifiers in the source reference, each identifier being associated with a document of a plurality of documents of the source shard, the hashing assigning each document of the plurality of documents to a target shard of the target shards, the plurality of documents being stored in a file associated with the source reference;

deleting at least some documents of the plurality of documents in the target references, the at least some documents belonging in a different target shard of the target shards;

hard linking the file into the target references;

marking the target index as read-write, such that the target index is used in place of the source index; and

deleting the source index.

2. The computer-implemented method of claim 1, wherein the target index has a same definition as the source index and a greater number of shards than the source index.

3. The computer-implemented method of claim 1, further comprising alternatively marking the source index as read-write when an error is reported by at least one of a file system and an application.

4. The computer-implemented method of claim 1, wherein the deleting denotes in the two or more target references that the at least some documents are deleted.

5. The computer-implemented method of claim 1, further comprising receiving a number target shards, the number of target shards being a factor of a number of source shards.

6. The computer-implemented method of claim 1, wherein the hash is a consistent hash.

7. The computer-implemented method of claim 1, wherein the source index and the target index are each an Elasticsearch index.

8. The computer-implemented method of claim 1, wherein the source reference and the two or more target references are each a Lucene index.

9. The computer implemented method of claim 1, wherein the source shard and the two or more target shards are all in a same data node.

10. The computer implemented method of claim 9, further comprising moving at least one of the two or more target shards to a different data node.

11. A system for splitting a shard comprising:

a processor; and

a memory coupled to the processor, the memory storing instructions executable by the processor to;

mark a source index as read only, the source index comprising a source shard, the source shard comprising a source reference;

create a target index, the target index comprising target shards, each target shard of the target shards comprising a target reference of target references;

copy the source reference, the copying producing the target references;

hash identifiers in the source reference, each identifier being associated with a document of a plurality of documents of the source shard, the hashing assigning each document of the plurality of documents to a target shard of the target shards, the plurality of documents being stored in a file associated with the source reference;

delete at least some documents of the plurality of documents in the target references, the at least some documents belonging in a different target shard of the target shards;

hard link the file into the target references;

mark the target index as read-write, such that the target index is used in place of the source index; and

delete the source index.

12. The system of claim 11, wherein the target index has a same definition as the source index and a greater number of shards than the source index.

13. The system of claim 11, wherein the instructions are further executable by the processor to alternatively mark the source index as read-write when an error is reported by at least one of a file system and an application.

14. The system of claim 11, wherein the deleting denotes in the two or more target references that the at least some documents are deleted.

15. The system of claim 11, wherein the instructions are further executable by the processor to receive a number target shards, the number of target shards being a factor of a number of source shards.

16. The system of claim 11, wherein the hash is a consistent hash.

17. The system of claim 11, wherein the source index and the target index are each an Elasticsearch index.

18. The system of claim 11, wherein the source reference and the two or more target references are each a Lucene index.

19. The system of claim 11, wherein

the source shard and the two or more target shards are all in a same data node, and

the instructions are further executable by the processor to move at least one of the two or more target shards to a different data node

20. A non-transitory computer-readable medium having embodied thereon a program, the program being executable by a processor to perform a method for splitting a shard, the method comprising:

marking a source index as read only, the source index comprising a source shard, the source shard comprising a source reference;

creating a target index, the target index comprising target shards, each target shard of the target shards comprising a target reference of target references;

copying the source reference, the copying producing the target references;

hashing identifiers in the source reference, each identifier being associated with a document of a plurality of documents of the source shard, the hashing assigning each document of the plurality of documents to a target shard of the target shards, the plurality of documents being stored in a file associated with the source reference;

deleting at least some documents of the plurality of documents in the target references, the at least some documents belonging in a different target shard of the target shards;

hard linking the file into the target references;

marking the target index as read-write, such that the target index is used in place of the source index; and

deleting the source index.