Skip to content

Commit a0c4cb1

Browse files
author
Devdutt Shenoi
committed
refactor: makes no sense to be Optional
1 parent 6916673 commit a0c4cb1

File tree

3 files changed

+31
-84
lines changed

3 files changed

+31
-84
lines changed

‎src/handlers/http/logstream.rs

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -651,8 +651,8 @@ pub async fn put_stream_hot_tier(
651651
let existing_hot_tier_used_size = hot_tier_manager
652652
.validate_hot_tier_size(&stream_name, &hottier.size)
653653
.await?;
654-
hottier.used_size = Some(existing_hot_tier_used_size.to_string());
655-
hottier.available_size = Some(hottier.size.clone());
654+
hottier.used_size = existing_hot_tier_used_size.to_string();
655+
hottier.available_size = hottier.size.clone();
656656
hottier.version = Some(CURRENT_HOT_TIER_VERSION.to_string());
657657
hot_tier_manager
658658
.put_hot_tier(&stream_name, &mut hottier)
@@ -695,8 +695,8 @@ pub async fn get_stream_hot_tier(req: HttpRequest) -> Result<impl Responder, Str
695695
if let Some(hot_tier_manager) = HotTierManager::global() {
696696
let mut hot_tier = hot_tier_manager.get_hot_tier(&stream_name).await?;
697697
hot_tier.size = format!("{} {}", hot_tier.size, "Bytes");
698-
hot_tier.used_size = Some(format!("{} {}", hot_tier.used_size.unwrap(), "Bytes"));
699-
hot_tier.available_size = Some(format!("{} {}", hot_tier.available_size.unwrap(), "Bytes"));
698+
hot_tier.used_size = format!("{} Bytes", hot_tier.used_size);
699+
hot_tier.available_size = format!("{} Bytes", hot_tier.available_size);
700700
Ok((web::Json(hot_tier), StatusCode::OK))
701701
} else {
702702
Err(StreamError::Custom {

‎src/hottier.rs

Lines changed: 21 additions & 71 deletions
Original file line numberDiff line numberDiff line change
@@ -56,10 +56,8 @@ pub const CURRENT_HOT_TIER_VERSION: &str = "v2";
5656
pub struct StreamHotTier {
5757
pub version: Option<String>,
5858
pub size: String,
59-
#[serde(skip_serializing_if = "Option::is_none")]
60-
pub used_size: Option<String>,
61-
#[serde(skip_serializing_if = "Option::is_none")]
62-
pub available_size: Option<String>,
59+
pub used_size: String,
60+
pub available_size: String,
6361
#[serde(skip_serializing_if = "Option::is_none")]
6462
pub oldest_date_time_entry: Option<String>,
6563
}
@@ -98,12 +96,7 @@ impl HotTierManager {
9896
if self.check_stream_hot_tier_exists(&stream) && stream != current_stream {
9997
let stream_hot_tier = self.get_hot_tier(&stream).await?;
10098
total_hot_tier_size += &stream_hot_tier.size.parse::<u64>().unwrap();
101-
total_hot_tier_used_size += &stream_hot_tier
102-
.used_size
103-
.clone()
104-
.unwrap()
105-
.parse::<u64>()
106-
.unwrap();
99+
total_hot_tier_used_size += stream_hot_tier.used_size.parse::<u64>().unwrap();
107100
}
108101
}
109102
Ok((total_hot_tier_size, total_hot_tier_used_size))
@@ -123,8 +116,7 @@ impl HotTierManager {
123116
if self.check_stream_hot_tier_exists(stream) {
124117
//delete existing hot tier if its size is less than the updated hot tier size else return error
125118
let existing_hot_tier = self.get_hot_tier(stream).await?;
126-
existing_hot_tier_used_size =
127-
existing_hot_tier.used_size.unwrap().parse::<u64>().unwrap();
119+
existing_hot_tier_used_size = existing_hot_tier.used_size.parse::<u64>().unwrap();
128120

129121
if stream_hot_tier_size < existing_hot_tier_used_size {
130122
return Err(HotTierError::ObjectStorageError(ObjectStorageError::Custom(format!(
@@ -260,12 +252,7 @@ impl HotTierManager {
260252
/// delete the files from the hot tier directory if the available date range is outside the hot tier range
261253
async fn process_stream(&self, stream: String) -> Result<(), HotTierError> {
262254
let stream_hot_tier = self.get_hot_tier(&stream).await?;
263-
let mut parquet_file_size = stream_hot_tier
264-
.used_size
265-
.as_ref()
266-
.unwrap()
267-
.parse::<u64>()
268-
.unwrap();
255+
let mut parquet_file_size = stream_hot_tier.used_size.parse::<u64>().unwrap();
269256

270257
let object_store = CONFIG.storage().get_object_store();
271258
let mut s3_manifest_file_list = object_store.list_manifest_files(&stream).await?;
@@ -357,13 +344,7 @@ impl HotTierManager {
357344
let mut file_processed = false;
358345
let mut stream_hot_tier = self.get_hot_tier(stream).await?;
359346
if !self.is_disk_available(parquet_file.file_size).await?
360-
|| stream_hot_tier
361-
.available_size
362-
.as_ref()
363-
.unwrap()
364-
.parse::<u64>()
365-
.unwrap()
366-
<= parquet_file.file_size
347+
|| stream_hot_tier.available_size.parse::<u64>().unwrap() <= parquet_file.file_size
367348
{
368349
if !self
369350
.cleanup_hot_tier_old_data(
@@ -376,12 +357,7 @@ impl HotTierManager {
376357
{
377358
return Ok(file_processed);
378359
}
379-
*parquet_file_size = stream_hot_tier
380-
.used_size
381-
.as_ref()
382-
.unwrap()
383-
.parse::<u64>()
384-
.unwrap();
360+
*parquet_file_size = stream_hot_tier.used_size.parse::<u64>().unwrap();
385361
}
386362
let parquet_file_path = RelativePathBuf::from(parquet_file.file_path.clone());
387363
fs::create_dir_all(parquet_path.parent().unwrap()).await?;
@@ -393,18 +369,11 @@ impl HotTierManager {
393369
.await?;
394370
file.write_all(&parquet_data).await?;
395371
*parquet_file_size += parquet_file.file_size;
396-
stream_hot_tier.used_size = Some(parquet_file_size.to_string());
397-
398-
stream_hot_tier.available_size = Some(
399-
(stream_hot_tier
400-
.available_size
401-
.as_ref()
402-
.unwrap()
403-
.parse::<u64>()
404-
.unwrap()
405-
- parquet_file.file_size)
406-
.to_string(),
407-
);
372+
stream_hot_tier.used_size = parquet_file_size.to_string();
373+
374+
stream_hot_tier.available_size = (stream_hot_tier.available_size.parse::<u64>().unwrap()
375+
- parquet_file.file_size)
376+
.to_string();
408377
self.put_hot_tier(stream, &mut stream_hot_tier).await?;
409378
file_processed = true;
410379
let mut hot_tier_manifest = self
@@ -614,35 +583,16 @@ impl HotTierManager {
614583
fs::remove_dir_all(path_to_delete.parent().unwrap()).await?;
615584
delete_empty_directory_hot_tier(path_to_delete.parent().unwrap()).await?;
616585

617-
stream_hot_tier.used_size = Some(
618-
(stream_hot_tier
619-
.used_size
620-
.as_ref()
621-
.unwrap()
622-
.parse::<u64>()
623-
.unwrap()
624-
- file_size)
625-
.to_string(),
626-
);
627-
stream_hot_tier.available_size = Some(
628-
(stream_hot_tier
629-
.available_size
630-
.as_ref()
631-
.unwrap()
632-
.parse::<u64>()
633-
.unwrap()
634-
+ file_size)
635-
.to_string(),
636-
);
586+
stream_hot_tier.used_size =
587+
(stream_hot_tier.used_size.parse::<u64>().unwrap() - file_size)
588+
.to_string();
589+
stream_hot_tier.available_size =
590+
(stream_hot_tier.available_size.parse::<u64>().unwrap() + file_size)
591+
.to_string();
637592
self.put_hot_tier(stream, stream_hot_tier).await?;
638593
delete_successful = true;
639594

640-
if stream_hot_tier
641-
.available_size
642-
.as_ref()
643-
.unwrap()
644-
.parse::<u64>()
645-
.unwrap()
595+
if stream_hot_tier.available_size.parse::<u64>().unwrap()
646596
<= parquet_file_size
647597
{
648598
continue 'loop_files;
@@ -740,8 +690,8 @@ impl HotTierManager {
740690
let mut stream_hot_tier = StreamHotTier {
741691
version: Some(CURRENT_HOT_TIER_VERSION.to_string()),
742692
size: INTERNAL_STREAM_HOT_TIER_SIZE_BYTES.to_string(),
743-
used_size: Some("0".to_string()),
744-
available_size: Some(INTERNAL_STREAM_HOT_TIER_SIZE_BYTES.to_string()),
693+
used_size: "0".to_string(),
694+
available_size: INTERNAL_STREAM_HOT_TIER_SIZE_BYTES.to_string(),
745695
oldest_date_time_entry: None,
746696
};
747697
self.put_hot_tier(INTERNAL_STREAM_NAME, &mut stream_hot_tier)

‎src/migration/mod.rs

Lines changed: 6 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -154,16 +154,13 @@ async fn migration_hot_tier(stream: &str) -> anyhow::Result<()> {
154154
stream_hot_tier.size = human_size_to_bytes(&stream_hot_tier.size)
155155
.unwrap()
156156
.to_string();
157-
stream_hot_tier.available_size = Some(
158-
human_size_to_bytes(&stream_hot_tier.available_size.unwrap())
157+
stream_hot_tier.available_size =
158+
human_size_to_bytes(&stream_hot_tier.available_size)
159159
.unwrap()
160-
.to_string(),
161-
);
162-
stream_hot_tier.used_size = Some(
163-
human_size_to_bytes(&stream_hot_tier.used_size.unwrap())
164-
.unwrap()
165-
.to_string(),
166-
);
160+
.to_string();
161+
stream_hot_tier.used_size = human_size_to_bytes(&stream_hot_tier.used_size)
162+
.unwrap()
163+
.to_string();
167164
hot_tier_manager
168165
.put_hot_tier(stream, &mut stream_hot_tier)
169166
.await?;

0 commit comments

Comments
 (0)