Skip to content

Commit b277775

Browse files
author
Devdutt Shenoi
authored
refactor: clean up parts of the codebase (#981)
1 parent 82a09eb commit b277775

24 files changed

+587
-703
lines changed

‎src/analytics.rs

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -64,7 +64,7 @@ pub struct Report {
6464
memory_total_bytes: u64,
6565
platform: String,
6666
storage_mode: String,
67-
server_mode: String,
67+
server_mode: Mode,
6868
version: String,
6969
commit_hash: String,
7070
active_ingestors: u64,
@@ -112,7 +112,7 @@ impl Report {
112112
memory_total_bytes: mem_total,
113113
platform: platform().to_string(),
114114
storage_mode: CONFIG.get_storage_mode_string().to_string(),
115-
server_mode: CONFIG.parseable.mode.to_string(),
115+
server_mode: CONFIG.parseable.mode,
116116
version: current().released_version.to_string(),
117117
commit_hash: current().commit_hash,
118118
active_ingestors: ingestor_metrics.0,

‎src/cli.rs

Lines changed: 6 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -527,20 +527,12 @@ impl FromArgMatches for Cli {
527527
.get_one::<usize>(Self::ROW_GROUP_SIZE)
528528
.cloned()
529529
.expect("default for row_group size");
530-
self.parquet_compression = match m
531-
.get_one::<String>(Self::PARQUET_COMPRESSION_ALGO)
532-
.expect("default for compression algo")
533-
.as_str()
534-
{
535-
"uncompressed" => Compression::UNCOMPRESSED,
536-
"snappy" => Compression::SNAPPY,
537-
"gzip" => Compression::GZIP,
538-
"lzo" => Compression::LZO,
539-
"brotli" => Compression::BROTLI,
540-
"lz4" => Compression::LZ4,
541-
"zstd" => Compression::ZSTD,
542-
_ => unreachable!(),
543-
};
530+
self.parquet_compression = serde_json::from_str(&format!(
531+
"{:?}",
532+
m.get_one::<String>(Self::PARQUET_COMPRESSION_ALGO)
533+
.expect("default for compression algo")
534+
))
535+
.expect("unexpected compression algo");
544536

545537
let openid_client_id = m.get_one::<String>(Self::OPENID_CLIENT_ID).cloned();
546538
let openid_client_secret = m.get_one::<String>(Self::OPENID_CLIENT_SECRET).cloned();

‎src/event/format/json.rs

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -45,11 +45,11 @@ impl EventFormat for Event {
4545
// also extract the arrow schema, tags and metadata from the incoming json
4646
fn to_data(
4747
self,
48-
schema: HashMap<String, Arc<Field>>,
49-
static_schema_flag: Option<String>,
50-
time_partition: Option<String>,
48+
schema: &HashMap<String, Arc<Field>>,
49+
static_schema_flag: Option<&String>,
50+
time_partition: Option<&String>,
5151
) -> Result<(Self::Data, Vec<Arc<Field>>, bool, Tags, Metadata), anyhow::Error> {
52-
let data = flatten_json_body(self.data, None, None, None, false)?;
52+
let data = flatten_json_body(&self.data, None, None, None, false)?;
5353
let stream_schema = schema;
5454

5555
// incoming event may be a single json or a json array
@@ -66,13 +66,13 @@ impl EventFormat for Event {
6666
collect_keys(value_arr.iter()).expect("fields can be collected from array of objects");
6767

6868
let mut is_first = false;
69-
let schema = match derive_arrow_schema(&stream_schema, fields) {
69+
let schema = match derive_arrow_schema(stream_schema, fields) {
7070
Ok(schema) => schema,
7171
Err(_) => match infer_json_schema_from_iterator(value_arr.iter().map(Ok)) {
7272
Ok(mut infer_schema) => {
7373
let new_infer_schema = super::super::format::update_field_type_in_schema(
7474
Arc::new(infer_schema),
75-
Some(&stream_schema),
75+
Some(stream_schema),
7676
time_partition,
7777
Some(&value_arr),
7878
);

‎src/event/format/mod.rs

Lines changed: 15 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -42,22 +42,21 @@ pub trait EventFormat: Sized {
4242

4343
fn to_data(
4444
self,
45-
schema: HashMap<String, Arc<Field>>,
46-
static_schema_flag: Option<String>,
47-
time_partition: Option<String>,
45+
schema: &HashMap<String, Arc<Field>>,
46+
static_schema_flag: Option<&String>,
47+
time_partition: Option<&String>,
4848
) -> Result<(Self::Data, EventSchema, bool, Tags, Metadata), AnyError>;
49+
4950
fn decode(data: Self::Data, schema: Arc<Schema>) -> Result<RecordBatch, AnyError>;
51+
5052
fn into_recordbatch(
5153
self,
52-
storage_schema: HashMap<String, Arc<Field>>,
53-
static_schema_flag: Option<String>,
54-
time_partition: Option<String>,
54+
storage_schema: &HashMap<String, Arc<Field>>,
55+
static_schema_flag: Option<&String>,
56+
time_partition: Option<&String>,
5557
) -> Result<(RecordBatch, bool), AnyError> {
56-
let (data, mut schema, is_first, tags, metadata) = self.to_data(
57-
storage_schema.clone(),
58-
static_schema_flag.clone(),
59-
time_partition.clone(),
60-
)?;
58+
let (data, mut schema, is_first, tags, metadata) =
59+
self.to_data(storage_schema, static_schema_flag, time_partition)?;
6160

6261
if get_field(&schema, DEFAULT_TAGS_KEY).is_some() {
6362
return Err(anyhow!("field {} is a reserved field", DEFAULT_TAGS_KEY));
@@ -120,8 +119,8 @@ pub trait EventFormat: Sized {
120119

121120
fn is_schema_matching(
122121
new_schema: Arc<Schema>,
123-
storage_schema: HashMap<String, Arc<Field>>,
124-
static_schema_flag: Option<String>,
122+
storage_schema: &HashMap<String, Arc<Field>>,
123+
static_schema_flag: Option<&String>,
125124
) -> bool {
126125
if static_schema_flag.is_none() {
127126
return true;
@@ -207,7 +206,7 @@ pub fn override_timestamp_fields(
207206
pub fn update_field_type_in_schema(
208207
inferred_schema: Arc<Schema>,
209208
existing_schema: Option<&HashMap<String, Arc<Field>>>,
210-
time_partition: Option<String>,
209+
time_partition: Option<&String>,
211210
log_records: Option<&Vec<Value>>,
212211
) -> Arc<Schema> {
213212
let mut updated_schema = inferred_schema.clone();
@@ -236,12 +235,12 @@ pub fn update_field_type_in_schema(
236235
if time_partition.is_none() {
237236
return updated_schema;
238237
}
239-
let time_partition_field_name = time_partition.unwrap();
238+
240239
let new_schema: Vec<Field> = updated_schema
241240
.fields()
242241
.iter()
243242
.map(|field| {
244-
if *field.name() == time_partition_field_name {
243+
if field.name() == time_partition.unwrap() {
245244
if field.data_type() == &DataType::Utf8 {
246245
let new_data_type = DataType::Timestamp(TimeUnit::Millisecond, None);
247246
Field::new(field.name().clone(), new_data_type, true)

‎src/handlers/http/ingest.rs

Lines changed: 17 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -61,7 +61,7 @@ pub async fn ingest(req: HttpRequest, body: Bytes) -> Result<HttpResponse, PostE
6161
}
6262
create_stream_if_not_exists(&stream_name, &StreamType::UserDefined.to_string()).await?;
6363

64-
flatten_and_push_logs(req, body, stream_name).await?;
64+
flatten_and_push_logs(req, body, &stream_name).await?;
6565
Ok(HttpResponse::Ok().finish())
6666
} else {
6767
Err(PostError::Header(ParseHeaderError::MissingStreamName))
@@ -84,7 +84,7 @@ pub async fn ingest_internal_stream(stream_name: String, body: Bytes) -> Result<
8484
tags: String::default(),
8585
metadata: String::default(),
8686
};
87-
event.into_recordbatch(schema, None, None)?
87+
event.into_recordbatch(&schema, None, None)?
8888
};
8989
event::Event {
9090
rb,
@@ -114,9 +114,9 @@ pub async fn handle_otel_ingestion(
114114
.iter()
115115
.find(|&(key, _)| key == STREAM_NAME_HEADER_KEY)
116116
{
117-
let stream_name = stream_name.to_str().unwrap().to_owned();
118-
create_stream_if_not_exists(&stream_name, &StreamType::UserDefined.to_string()).await?;
119-
push_logs(stream_name.to_string(), req.clone(), body).await?;
117+
let stream_name = stream_name.to_str().unwrap();
118+
create_stream_if_not_exists(stream_name, &StreamType::UserDefined.to_string()).await?;
119+
push_logs(stream_name, &req, &body).await?;
120120
} else {
121121
return Err(PostError::Header(ParseHeaderError::MissingStreamName));
122122
}
@@ -149,7 +149,7 @@ pub async fn post_event(req: HttpRequest, body: Bytes) -> Result<HttpResponse, P
149149
}
150150
}
151151

152-
flatten_and_push_logs(req, body, stream_name).await?;
152+
flatten_and_push_logs(req, body, &stream_name).await?;
153153
Ok(HttpResponse::Ok().finish())
154154
}
155155

@@ -319,7 +319,7 @@ mod tests {
319319
.append_header((PREFIX_META.to_string() + "C", "meta1"))
320320
.to_http_request();
321321

322-
let (rb, _) = into_event_batch(req, json, HashMap::default(), None, None).unwrap();
322+
let (rb, _) = into_event_batch(&req, &json, HashMap::default(), None, None).unwrap();
323323

324324
assert_eq!(rb.num_rows(), 1);
325325
assert_eq!(rb.num_columns(), 6);
@@ -359,7 +359,7 @@ mod tests {
359359

360360
let req = TestRequest::default().to_http_request();
361361

362-
let (rb, _) = into_event_batch(req, json, HashMap::default(), None, None).unwrap();
362+
let (rb, _) = into_event_batch(&req, &json, HashMap::default(), None, None).unwrap();
363363

364364
assert_eq!(rb.num_rows(), 1);
365365
assert_eq!(rb.num_columns(), 5);
@@ -391,7 +391,7 @@ mod tests {
391391

392392
let req = TestRequest::default().to_http_request();
393393

394-
let (rb, _) = into_event_batch(req, json, schema, None, None).unwrap();
394+
let (rb, _) = into_event_batch(&req, &json, schema, None, None).unwrap();
395395

396396
assert_eq!(rb.num_rows(), 1);
397397
assert_eq!(rb.num_columns(), 5);
@@ -423,7 +423,7 @@ mod tests {
423423

424424
let req = TestRequest::default().to_http_request();
425425

426-
assert!(into_event_batch(req, json, schema, None, None).is_err());
426+
assert!(into_event_batch(&req, &json, schema, None, None).is_err());
427427
}
428428

429429
#[test]
@@ -441,7 +441,7 @@ mod tests {
441441

442442
let req = TestRequest::default().to_http_request();
443443

444-
let (rb, _) = into_event_batch(req, json, schema, None, None).unwrap();
444+
let (rb, _) = into_event_batch(&req, &json, schema, None, None).unwrap();
445445

446446
assert_eq!(rb.num_rows(), 1);
447447
assert_eq!(rb.num_columns(), 3);
@@ -453,7 +453,7 @@ mod tests {
453453

454454
let req = TestRequest::default().to_http_request();
455455

456-
assert!(into_event_batch(req, json, HashMap::default(), None, None).is_err())
456+
assert!(into_event_batch(&req, &json, HashMap::default(), None, None).is_err())
457457
}
458458

459459
#[test]
@@ -476,7 +476,7 @@ mod tests {
476476

477477
let req = TestRequest::default().to_http_request();
478478

479-
let (rb, _) = into_event_batch(req, json, HashMap::default(), None, None).unwrap();
479+
let (rb, _) = into_event_batch(&req, &json, HashMap::default(), None, None).unwrap();
480480

481481
assert_eq!(rb.num_rows(), 3);
482482
assert_eq!(rb.num_columns(), 6);
@@ -524,7 +524,7 @@ mod tests {
524524

525525
let req = TestRequest::default().to_http_request();
526526

527-
let (rb, _) = into_event_batch(req, json, HashMap::default(), None, None).unwrap();
527+
let (rb, _) = into_event_batch(&req, &json, HashMap::default(), None, None).unwrap();
528528

529529
assert_eq!(rb.num_rows(), 3);
530530
assert_eq!(rb.num_columns(), 6);
@@ -572,7 +572,7 @@ mod tests {
572572
);
573573
let req = TestRequest::default().to_http_request();
574574

575-
let (rb, _) = into_event_batch(req, json, schema, None, None).unwrap();
575+
let (rb, _) = into_event_batch(&req, &json, schema, None, None).unwrap();
576576

577577
assert_eq!(rb.num_rows(), 3);
578578
assert_eq!(rb.num_columns(), 6);
@@ -621,7 +621,7 @@ mod tests {
621621
.into_iter(),
622622
);
623623

624-
assert!(into_event_batch(req, json, schema, None, None).is_err());
624+
assert!(into_event_batch(&req, &json, schema, None, None).is_err());
625625
}
626626

627627
#[test]
@@ -649,7 +649,7 @@ mod tests {
649649

650650
let req = TestRequest::default().to_http_request();
651651

652-
let (rb, _) = into_event_batch(req, json, HashMap::default(), None, None).unwrap();
652+
let (rb, _) = into_event_batch(&req, &json, HashMap::default(), None, None).unwrap();
653653

654654
assert_eq!(rb.num_rows(), 4);
655655
assert_eq!(rb.num_columns(), 7);

‎src/handlers/http/kinesis.rs

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -24,16 +24,14 @@ use std::collections::BTreeMap;
2424
use std::str;
2525

2626
#[derive(Serialize, Deserialize, Debug)]
27+
#[serde(rename_all = "camelCase")]
2728
struct Message {
28-
#[serde(rename = "records")]
2929
records: Vec<Data>,
30-
#[serde(rename = "requestId")]
3130
request_id: String,
3231
timestamp: u64,
3332
}
3433
#[derive(Serialize, Deserialize, Debug)]
3534
struct Data {
36-
#[serde(rename = "data")]
3735
data: String,
3836
}
3937

‎src/handlers/http/logstream.rs

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -614,8 +614,8 @@ pub async fn put_stream_hot_tier(
614614
let existing_hot_tier_used_size = hot_tier_manager
615615
.validate_hot_tier_size(&stream_name, &hottier.size)
616616
.await?;
617-
hottier.used_size = Some(existing_hot_tier_used_size.to_string());
618-
hottier.available_size = Some(hottier.size.clone());
617+
hottier.used_size = existing_hot_tier_used_size.to_string();
618+
hottier.available_size = hottier.size.to_string();
619619
hottier.version = Some(CURRENT_HOT_TIER_VERSION.to_string());
620620
hot_tier_manager
621621
.put_hot_tier(&stream_name, &mut hottier)
@@ -658,8 +658,8 @@ pub async fn get_stream_hot_tier(req: HttpRequest) -> Result<impl Responder, Str
658658
if let Some(hot_tier_manager) = HotTierManager::global() {
659659
let mut hot_tier = hot_tier_manager.get_hot_tier(&stream_name).await?;
660660
hot_tier.size = format!("{} {}", hot_tier.size, "Bytes");
661-
hot_tier.used_size = Some(format!("{} {}", hot_tier.used_size.unwrap(), "Bytes"));
662-
hot_tier.available_size = Some(format!("{} {}", hot_tier.available_size.unwrap(), "Bytes"));
661+
hot_tier.used_size = format!("{} Bytes", hot_tier.used_size);
662+
hot_tier.available_size = format!("{} Bytes", hot_tier.available_size);
663663
Ok((web::Json(hot_tier), StatusCode::OK))
664664
} else {
665665
Err(StreamError::Custom {

‎src/handlers/http/middleware.rs

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -44,8 +44,8 @@ use crate::{
4444
use serde::{Deserialize, Serialize};
4545

4646
#[derive(Serialize, Deserialize, Debug)]
47+
#[serde(rename_all = "camelCase")]
4748
struct Message {
48-
#[serde(rename = "commonAttributes")]
4949
common_attributes: CommonAttributes,
5050
}
5151

‎src/handlers/http/modal/ingest_server.rs

Lines changed: 2 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -43,7 +43,6 @@ use crate::storage::PARSEABLE_ROOT_DIRECTORY;
4343
use crate::sync;
4444

4545
use crate::{handlers::http::base_path, option::CONFIG};
46-
use actix_web::body::MessageBody;
4746
use actix_web::web;
4847
use actix_web::web::resource;
4948
use actix_web::Scope;
@@ -309,9 +308,7 @@ impl IngestServer {
309308
.clone_from(&INGESTOR_META.domain_name);
310309
store_data.port.clone_from(&INGESTOR_META.port);
311310

312-
let resource = serde_json::to_string(&store_data)?
313-
.try_into_bytes()
314-
.map_err(|err| anyhow!(err))?;
311+
let resource = Bytes::from(serde_json::to_vec(&store_data)?);
315312

316313
// if pushing to object store fails propagate the error
317314
return store
@@ -320,9 +317,7 @@ impl IngestServer {
320317
.map_err(|err| anyhow!(err));
321318
}
322319
} else {
323-
let resource = serde_json::to_string(&resource)?
324-
.try_into_bytes()
325-
.map_err(|err| anyhow!(err))?;
320+
let resource = Bytes::from(serde_json::to_vec(&resource)?);
326321

327322
store.put_object(&path, resource).await?;
328323
}

‎src/handlers/http/modal/mod.rs

Lines changed: 2 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -221,6 +221,7 @@ impl IngestorMetadata {
221221
#[cfg(test)]
222222
mod test {
223223
use actix_web::body::MessageBody;
224+
use bytes::Bytes;
224225
use rstest::rstest;
225226

226227
use super::{IngestorMetadata, DEFAULT_VERSION};
@@ -256,10 +257,7 @@ mod test {
256257
"8002".to_string(),
257258
);
258259

259-
let lhs = serde_json::to_string(&im)
260-
.unwrap()
261-
.try_into_bytes()
262-
.unwrap();
260+
let lhs = Bytes::from(serde_json::to_vec(&im).unwrap());
263261
let rhs = br#"{"version":"v3","port":"8000","domain_name":"https://localhost:8000","bucket_name":"somebucket","token":"Basic YWRtaW46YWRtaW4=","ingestor_id":"ingestor_id","flight_port":"8002"}"#
264262
.try_into_bytes()
265263
.unwrap();

0 commit comments

Comments
 (0)