-
Notifications
You must be signed in to change notification settings - Fork 102
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Consolidate Index and TimeIndex into single Index
- Loading branch information
Showing
44 changed files
with
248 additions
and
1,288 deletions.
There are no files selected for viewing
Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,94 @@ | ||
| use crate::streaming::utils::file; | ||
| use crate::{server_error::ServerError, streaming::segments::storage::INDEX_SIZE}; | ||
| use bytes::{BufMut, BytesMut}; | ||
| use tokio::io::{AsyncReadExt, AsyncWriteExt, BufReader, BufWriter}; | ||
| use tracing::trace; | ||
|
|
||
| pub struct IndexConverter { | ||
| pub index_path: String, | ||
| pub time_index_path: String, | ||
| } | ||
|
|
||
| impl IndexConverter { | ||
| pub fn new(index_path: String, time_index_path: String) -> Self { | ||
| Self { | ||
| index_path, | ||
| time_index_path, | ||
| } | ||
| } | ||
|
|
||
| pub async fn migrate(&self) -> Result<(), ServerError> { | ||
| let indexes = self.convert_indexes().await?; | ||
| self.replace_with_converted(indexes).await?; | ||
|
|
||
| Ok(()) | ||
| } | ||
|
|
||
| pub async fn needs_migration(&self) -> Result<bool, ServerError> { | ||
| Ok(file::exists(&self.time_index_path).await?) | ||
| } | ||
|
|
||
| async fn convert_indexes(&self) -> Result<Vec<u8>, ServerError> { | ||
| let index_file = file::open(&self.index_path).await?; | ||
| let time_index_file = file::open(&self.time_index_path).await?; | ||
|
|
||
| let mut index_reader = BufReader::new(index_file); | ||
| let mut time_index_reader = BufReader::new(time_index_file); | ||
|
|
||
| let new_index = Vec::new(); | ||
| let mut new_index_writer = BufWriter::new(new_index); | ||
|
|
||
| loop { | ||
| let relative_offset_result = index_reader.read_u32_le().await; | ||
| let position_result = index_reader.read_u32_le().await; | ||
|
|
||
| let time_relative_offset_result = time_index_reader.read_u32_le().await; | ||
| let timestamp_result = time_index_reader.read_u64_le().await; | ||
|
|
||
| if relative_offset_result.is_err() | ||
| || position_result.is_err() | ||
| || time_relative_offset_result.is_err() | ||
| || timestamp_result.is_err() | ||
| { | ||
| trace!("Reached EOF for index file: {}", &self.index_path); | ||
| break; | ||
| } | ||
|
|
||
| let relative_offset = relative_offset_result?; | ||
| let position = position_result?; | ||
| let time_relative_offset = time_relative_offset_result?; | ||
| let timestamp = timestamp_result?; | ||
|
|
||
| if relative_offset != time_relative_offset { | ||
| return Err(ServerError::IndexMigrationError( | ||
| "Mismatched relative offsets in normal index: {relative_offset} vs time index: {time_relative_offset}".to_string(), | ||
| )); | ||
| } | ||
|
|
||
| let mut new_index_entry = BytesMut::with_capacity(INDEX_SIZE as usize); // u32 + u32 + u64 | ||
| new_index_entry.put_u32_le(relative_offset); | ||
| new_index_entry.put_u32_le(position); | ||
| new_index_entry.put_u64_le(timestamp); | ||
|
|
||
| new_index_writer.write_all(&new_index_entry).await?; | ||
| } | ||
| new_index_writer.flush().await?; | ||
|
|
||
| Ok(new_index_writer.into_inner()) | ||
| } | ||
|
|
||
| async fn replace_with_converted(&self, indexes: Vec<u8>) -> Result<(), ServerError> { | ||
| let _ = file::remove(&self.index_path).await; | ||
| let _ = file::remove(&self.time_index_path).await; | ||
|
|
||
| let new_index_file = file::overwrite(&self.index_path).await?; | ||
|
|
||
| let mut new_index_writer = BufWriter::new(new_index_file); | ||
| new_index_writer.write_all(&indexes).await?; | ||
| new_index_writer.flush().await?; | ||
|
|
||
| trace!("Replaced old index with new index at {}", &self.index_path); | ||
|
|
||
| Ok(()) | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1 @@ | ||
| pub mod index_converter; |
This file was deleted.
Oops, something went wrong.
This file was deleted.
Oops, something went wrong.
Oops, something went wrong.