Skip to content

Commit 0138d5c

Browse files
author
Devdutt Shenoi
committed
refactor: non-optionalize params
1 parent fc9e0ed commit 0138d5c

File tree

2 files changed

+22
-23
lines changed

2 files changed

+22
-23
lines changed

‎server/src/handlers/http/logstream.rs

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -566,8 +566,8 @@ pub async fn put_stream_hot_tier(
566566
let existing_hot_tier_used_size = hot_tier_manager
567567
.validate_hot_tier_size(&stream_name, hottier.size)
568568
.await?;
569-
hottier.used_size = Some(existing_hot_tier_used_size);
570-
hottier.available_size = Some(hottier.size.clone());
569+
hottier.used_size = existing_hot_tier_used_size;
570+
hottier.available_size = hottier.size;
571571
hottier.version = CURRENT_HOT_TIER_VERSION.to_string();
572572
hot_tier_manager
573573
.put_hot_tier(&stream_name, &mut hottier)
@@ -601,8 +601,8 @@ pub async fn get_stream_hot_tier(req: HttpRequest) -> Result<impl Responder, Str
601601
let hot_tier = hot_tier_manager.get_hot_tier(&stream_name).await?;
602602
let json = serde_json::json!({
603603
"size": format!("{} Bytes", hot_tier.size),
604-
"used_size": format!("{} Bytes", hot_tier.used_size.unwrap()),
605-
"available_size": format!("{} Bytes", hot_tier.available_size.unwrap()),
604+
"used_size": format!("{} Bytes", hot_tier.used_size),
605+
"available_size": format!("{} Bytes", hot_tier.available_size),
606606
});
607607
Ok((web::Json(json), StatusCode::OK))
608608
} else {

‎server/src/hottier.rs

Lines changed: 18 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -60,8 +60,8 @@ pub const CURRENT_HOT_TIER_VERSION: &str = "v2";
6060
pub struct StreamHotTier {
6161
pub version: String,
6262
pub size: u64,
63-
pub used_size: Option<u64>,
64-
pub available_size: Option<u64>,
63+
pub used_size: u64,
64+
pub available_size: u64,
6565
pub oldest_date_time_entry: Option<String>,
6666
}
6767

@@ -99,7 +99,7 @@ impl HotTierManager {
9999
if self.check_stream_hot_tier_exists(&stream) && stream != current_stream {
100100
let stream_hot_tier = self.get_hot_tier(&stream).await?;
101101
total_hot_tier_size += stream_hot_tier.size;
102-
total_hot_tier_used_size += stream_hot_tier.used_size.as_ref().unwrap();
102+
total_hot_tier_used_size += stream_hot_tier.used_size;
103103
}
104104
}
105105
Ok((total_hot_tier_size, total_hot_tier_used_size))
@@ -118,7 +118,7 @@ impl HotTierManager {
118118
if self.check_stream_hot_tier_exists(stream) {
119119
//delete existing hot tier if its size is less than the updated hot tier size else return error
120120
let existing_hot_tier = self.get_hot_tier(stream).await?;
121-
existing_hot_tier_used_size = *existing_hot_tier.used_size.as_ref().unwrap();
121+
existing_hot_tier_used_size = existing_hot_tier.used_size;
122122

123123
if stream_hot_tier_size < existing_hot_tier_used_size {
124124
return Err(HotTierError::ObjectStorageError(ObjectStorageError::Custom(format!(
@@ -188,11 +188,13 @@ impl HotTierManager {
188188
used_size: stream_hot_tier
189189
.used_size
190190
.as_ref()
191-
.map(|size| human_size_to_bytes(size).unwrap()),
191+
.map(|size| human_size_to_bytes(size).unwrap())
192+
.unwrap_or_default(),
192193
available_size: stream_hot_tier
193194
.available_size
194195
.as_ref()
195-
.map(|size| human_size_to_bytes(size).unwrap()),
196+
.map(|size| human_size_to_bytes(size).unwrap())
197+
.unwrap_or_default(),
196198
oldest_date_time_entry: stream_hot_tier.oldest_date_time_entry,
197199
};
198200

@@ -276,7 +278,7 @@ impl HotTierManager {
276278
/// delete the files from the hot tier directory if the available date range is outside the hot tier range
277279
async fn process_stream(&self, stream: String) -> Result<(), HotTierError> {
278280
let stream_hot_tier = self.get_hot_tier(&stream).await?;
279-
let mut parquet_file_size = *stream_hot_tier.used_size.as_ref().unwrap();
281+
let mut parquet_file_size = stream_hot_tier.used_size;
280282

281283
let object_store = CONFIG.storage().get_object_store();
282284
let mut s3_manifest_file_list = object_store.list_manifest_files(&stream).await?;
@@ -368,7 +370,7 @@ impl HotTierManager {
368370
let mut file_processed = false;
369371
let mut stream_hot_tier = self.get_hot_tier(stream).await?;
370372
if !self.is_disk_available(parquet_file.file_size).await?
371-
|| stream_hot_tier.available_size.as_ref().unwrap() <= &parquet_file.file_size
373+
|| stream_hot_tier.available_size <= parquet_file.file_size
372374
{
373375
if !self
374376
.cleanup_hot_tier_old_data(
@@ -381,7 +383,7 @@ impl HotTierManager {
381383
{
382384
return Ok(file_processed);
383385
}
384-
*parquet_file_size = *stream_hot_tier.used_size.as_ref().unwrap();
386+
*parquet_file_size = stream_hot_tier.used_size;
385387
}
386388
let parquet_file_path = RelativePathBuf::from(parquet_file.file_path.clone());
387389
fs::create_dir_all(parquet_path.parent().unwrap()).await?;
@@ -393,10 +395,9 @@ impl HotTierManager {
393395
.await?;
394396
file.write_all(&parquet_data).await?;
395397
*parquet_file_size += parquet_file.file_size;
396-
stream_hot_tier.used_size = Some(*parquet_file_size);
398+
stream_hot_tier.used_size = *parquet_file_size;
397399

398-
stream_hot_tier.available_size =
399-
Some(stream_hot_tier.available_size.as_ref().unwrap() - parquet_file.file_size);
400+
stream_hot_tier.available_size -= parquet_file.file_size;
400401
self.put_hot_tier(stream, &mut stream_hot_tier).await?;
401402
file_processed = true;
402403
let mut hot_tier_manifest = self
@@ -606,14 +607,12 @@ impl HotTierManager {
606607
fs::remove_dir_all(path_to_delete.parent().unwrap()).await?;
607608
delete_empty_directory_hot_tier(path_to_delete.parent().unwrap()).await?;
608609

609-
stream_hot_tier.used_size =
610-
Some(stream_hot_tier.used_size.as_ref().unwrap() - file_size);
611-
stream_hot_tier.available_size =
612-
Some(stream_hot_tier.available_size.as_ref().unwrap() + file_size);
610+
stream_hot_tier.used_size -= file_size;
611+
stream_hot_tier.available_size += file_size;
613612
self.put_hot_tier(stream, stream_hot_tier).await?;
614613
delete_successful = true;
615614

616-
if stream_hot_tier.available_size.as_ref().unwrap() <= &parquet_file_size {
615+
if stream_hot_tier.available_size <= parquet_file_size {
617616
continue 'loop_files;
618617
} else {
619618
break 'loop_dates;
@@ -709,8 +708,8 @@ impl HotTierManager {
709708
let mut stream_hot_tier = StreamHotTier {
710709
version: CURRENT_HOT_TIER_VERSION.to_string(),
711710
size: INTERNAL_STREAM_HOT_TIER_SIZE_BYTES,
712-
used_size: Some(0),
713-
available_size: Some(INTERNAL_STREAM_HOT_TIER_SIZE_BYTES),
711+
used_size: 0,
712+
available_size: INTERNAL_STREAM_HOT_TIER_SIZE_BYTES,
714713
oldest_date_time_entry: None,
715714
};
716715
self.put_hot_tier(INTERNAL_STREAM_NAME, &mut stream_hot_tier)

0 commit comments

Comments
 (0)