[jira] [Assigned] (IGNITE-17990) Download RAFT snapshot without deleting original partition data

["Kirill Tkalenko (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/IGNITE-17990?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Kirill Tkalenko reassigned IGNITE-17990:
----------------------------------------

    Assignee: Kirill Tkalenko

> Download RAFT snapshot without deleting original partition data
> ---------------------------------------------------------------
>
>                 Key: IGNITE-17990
>                 URL: https://issues.apache.org/jira/browse/IGNITE-17990
>             Project: Ignite
>          Issue Type: Improvement
>            Reporter: Ivan Bessonov
>            Assignee: Kirill Tkalenko
>            Priority: Major
>              Labels: ignite-3
>             Fix For: 3.0.0-beta2
>
>
> h3. Note
> This is an umbrella issue. It's split into several smaller issues.
> {color:red}See the first comment, it indicates how we end up doing a full rebalancing.{color}
> h3. Description
> In first design, full rebalance is implemented this way:
>  * we drop partition data
>  * we download partition data from the leader
>  * we're done
> There's a problem with this approach - if download part failed, we lost one follower. This is bad, because technically new leader may have more data in the log and it could have uploaded it the follower, but now it makes no sense.
> Not only can it lead to hard-to-catch errors and introducing custom work-around code to JRaft, it's also an unconditional data deletion without neither explicit  user approval nor a copy of the data preserved durably.
> Such implementation is fine for POC and some tests, but it cannot be allowed in the release version of the product.
> h3. New proposed solution
> As trivial as it may seem, new solution is to _not deleting data_ before snapshot is fully downloaded and ready for swap. Why is it trivial? Because this is literally what RAFT demands to be done.
> Of course, there's a {*}but{*}. Snapshot application, when it's downloaded, should be {{O(1)}} when it comes to the number of rows in the partition and a number of transactions in a tx state store. This may not be fully achievable, depending on the implementation that we chose, more on that later.
> Following sections will describe all my concerns and possible implementations. Some sections can be skipped while reading. For example, if you're not interested in a specific storage engine, but want to read everything else.
> h3. TX state storage
> There's one really good thing about TX state storage. It has no storage engine, there's only a single RocksDB-based implementation. This makes possible the following approach:
>  * when we stream data, we can write it into a SST file, almost like in snapshots of meta-storage and CMG storages
>  * once snapshot is downloaded, we ingest it into a storage
> What I like about this solution is that it's very simple. But, there are concerns:
>  * ingesting leads to additional implicit data flush. Maybe it can be avoided, more on that later
>  * it's not clear whether RocksDB creates a copy of SST file or not. I would assume that it does, because the file might be in other folder or on another device, for example. Although copying files is fast, it still takes time. Add to this a time required for the flush and we see a problem - operation may become unnecessarily long
> For these reasons, I don't think that such solution should be implemented. The point of this description was to show, that I thought about this alternative and consciously decided to use another one.
> I believe that TX state storage should use the same approach as a RocksDB-based partition storage. Its description can be found later in this issue.
> h3. MV storage - Test engine
> Test uses concurrent skip-list map for MV data and a bunch of other maps for indexes. While snapshots is being downloaded, we should insert all data into new maps, that have the same structure. In the end, we should have two versions of the partition: old and new.
> {{onSnapshotLoad}} should just swap all objects. After that, old partition data can be cleaned by the garbage collector.
> This is a good place to start implementation. I assume that some new API will be introduced. I have thoughts about it as well, they are described later in *API* section.
> h3. MV storage - RocksDB engine
> SST-based approach is described in a *TX state storage* section. There I describe why I don't think that this is a good solution. Same reasoning can be applied here just as effectively. This means that we should write data in the same RocksDB instance. This is a little bit tricky.
> The reason is that all stored data is merged together, and Columns Families are shared between different partitions. This makes it harder to find a place to write partition data while old partition data persists. As a reminder and an example, let's take a look at how data is stored in row storage:
> {code:java}
> +-------------------+-----+-------------------+
> |    Partition 0    | ... |   Partition MAX   |
> +-------------------+-----+-------------------+
> | Row1 | ... | RowN | ... | Row1 | ... | RowN |
> +-------------------+-----+-------------------+{code}
> Logically, CF is split into a different "blocks", and each block represents a partition. Currently, each partition block is associated with an 2-bytes identifier that matches a partition number in Big Endian.
>  
> We could add new CF with similar structure and write snapshot data in it, but then the snapshot load operation would require us to move data from one CF to another. The only option that I know of, that can do this, is SST ingestion. And I already explained why I don't like it.
> This leaves us with the necessity to write data into the same column family. Naturally occurring solution is to assign a new identifier to the "new" version of partition. This way replacing "old" partition with "new" would be implemented by replacing "oldPartId" to "newPartId" in table storage metadata.
> Sounds good. No flush is required, snapshot loading becomes pretty fast.
> The only thing to keep in mind is that there are multiple column families in each partition - row data, hash indexes and a CF for every sorted index.
> When "old" partition is deleted, we should probably somehow hint that RocksDB should merge some layers and remove a substantial amount of data from disk. But such optimization also relates to general partition eviction and is out of scope of the discussion.
> Last point: what is "oldPartId" and "newPartId"?
> Logically, partition id now becomes a tuple of partition number and its generation. Physically it should be represented as an integer, where lower bits are generation and higher bits are partition number. It it 2 or 3 bytes? Good question, I'll answer it later.
> {code:java}
> | Partition Number | Generation |{code}
> Every time we need a rebalance, we increase the generation of the current partition. Generation counter overflows without affecting the partition number.
>  
> There are alternatives - we either have N possible generations (rotating, of course) or only 2 (0 and 1). Why is this important?
> Every time a partition is "started", it should, technically, perform a cleanup. Imagine (for example) we have partition {{{}(23,g){}}}. Then we would have to cleanup following ranges (lower bound inclusive, upper bound exclusive):
>  * for 256 generations - ranges {{(23,0):(23,g)}} and {{(23,g+1):(24,0)}}
>  * for 2 generations - range {{(23,1-g):(23,2-g)}}
> This should be done for all indexes as well. Something tells me that recovery will be somewhat faster in the second case, but is it more convenient? I don't know.
> Lastly, random thought on the prefix size. Don't we increase a footprint by storing it in every key? Yes and no. Right now the engine is not yet properly tuned, but in the future, we may set it up in such a way that RocksDB trims prefixes from the keys, so the prefix is kind of irrelevant. Will we configure a prefix to be a partition id or a pair (partId, rowId) - I don't know yet. Both options look good, but second may be better. We'll should do both and benchmark them.
> h3. MV storage - Persistent Page Memory
> This engine is way less complicated, but there are tricky parts as well. First of all, we can't have new partition ids like in RocksDB engine. It's 2 bytes, period. Too much code depends on this size to be exactly two bytes. There's a possibility of reducing a maximum number of partitions to a 32 thousands or so, leaving us with a single "free" bit to store generation, like in proposed RocksDB implementation. 
> Second, unlike RocksDB, each partition is stored in its own set of files. Or, in other words, partitions are completely independent, which greatly simplifies the swapping procedure. I propose the following algorithm:
>  * create new partition generation and upload data into it
>  * checkpoint everything
>  * invalidate all partition pages in page memory (this is {{{}O(1){}}}) for the old generation and close all file descriptors (page-stores)
>  * drop new partition files
> Local storage recovery should be implemented carefully. I propose having a rebalance marker, that shows that rebalance is in progress. On recovery:
>  * if marker is present, delete "new" partition files
>  * if marker is absent, do nothing
> Marker should contain the generation information, otherwise there might be a confusion. Real generation must be easy to determine.
> h3. MV storage - Volatile Page Memory
> In case of a volatile storage, there are no files to work with and there's even no need to partition generations. Instead, we can preserve old partition "meta" - pointers to all trees, free lists or anything else useful. After that, we start writing data into a new partition with a new meta. Pretty much like in Test storage engine, but offheap.
> When we're done, we can start deleting old partition data, using the same mechanism that's used in partition eviction (IGNITE-17833, not implemented yet).
> If snapshot downloading is interrupted, we delete everything that we already downloaded asynchronously, again, reusing the partition eviction code. No memory leaks should be left from it.
> h3. Atomic snapshot load in case of multiple storages
> On every step, every operation may fail. But data consistency should be preserved no matter what. Here, in particular, we need to call two {{onSnapshotLoad}} methods atomically. Their implementations may be very different.
> On high level, operation may look like this:
>  * write operation marker somewhere (table folder, vault, doesn't matter). Before doing so, we need to make sure that data is persisted on disk for both storages. Once marker is created, there's no way back. Old partition data will be destroyed
>  * call both methods
>  * remove marker when load is completed
> Pretty simple. Just do the same thing on recovery, if marker is present. One thing to keep in mind - {{onSnapshotLoad}} should be idempotent for this to work. If new partition is already loaded, nothing should break. Loading should effectively become a no-op in such case.
> h3. API
> I realize that MvPartitionStorage interface slowly becomes a mess. There's not much that we can do with it. But, rebalance is a good exception to the rule.
> Basically, we can implement something like this:
> {code:java}
> void performLocalRecovery();
> MvRebalanceDataWriter startDataRebalancing();
> interface MvRebalanceDataWriter {
>     CompletableFuture<?> beforeWritingData();
>     void addCommitted(...);
>     void addUncommitted(...); // Doesn't return the old value, because that's actually pointless during rebalancing
>     // lastAppliedIndex, runConsistently, etc.
>     CompletableFuture<?> afterWritingData();
>     void close();
> }{code}
> What exactly do we do in {{onSnapshotLoad}} and which interfaces it uses, I'm not sure at the moment. I just hope that this brief description gives you a gist of what I would like to see in the code. I do believe that it will simplify the API at least a little bit.
> What about indexes? That's a good question. I would expect that old objects, created with {{{}getOrCreate*Index{}}}, should still be functional. It may be a pain in the rear, we may have to introduce default implementation with changeable delegates. It's hard to predict exactly, but this is definitely a part that also requires attention. Same applies to partitions, actually.
> h3. Read Access
> One of the last things to discuss is a possibility to read data while rebalance is in the process. As we know, "full rebalance" can theoretically be performed on a live follower. RAFT works around this fact by allowing reading data from leader only. We don't have such limitation. Instead we have problems.
> Technically, it's very easy to allow reads until the {{onSnapshotLoad}} happens. Everything's simple when it's completed also. Any read operation during onSnapshotLoad should wait for its completion.
> That's good for "fast" read operations. But we also have scans. There are two options:
>  * we cancel them
>  * we allow them continue working seamlessly
> I think the second option is better. When cursor is notified that the "switch" is happening, it opens a new underlying cursor from the position that it previously stopped on. It feels like each storage will have its own version, they may have small differences.
> h3. Conclusion
> I know that this is a pretty big task. I don't expect it to be done in one sitting. It should be split to 3 issues at least. Probably more.
> This is fine, just don't forget the link to this particular issue, because it has the overall description of what's going on.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)