Skip to content

Commit 8dee91a

Browse files
author
Devdutt Shenoi
committed
feat: don't clone on events
1 parent fcd6988 commit 8dee91a

File tree

4 files changed

+26
-23
lines changed

4 files changed

+26
-23
lines changed

src/event/format/json.rs

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -45,9 +45,9 @@ 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> {
5252
let data = flatten_json_body(self.data, None, None, None, false)?;
5353
let stream_schema = schema;
@@ -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;
@@ -225,7 +224,7 @@ pub fn override_num_fields_from_schema(schema: Vec<Arc<Field>>) -> Vec<Arc<Field
225224
pub fn update_field_type_in_schema(
226225
inferred_schema: Arc<Schema>,
227226
existing_schema: Option<&HashMap<String, Arc<Field>>>,
228-
time_partition: Option<String>,
227+
time_partition: Option<&String>,
229228
log_records: Option<&Vec<Value>>,
230229
) -> Arc<Schema> {
231230
let mut updated_schema = inferred_schema.clone();
@@ -258,12 +257,12 @@ pub fn update_field_type_in_schema(
258257
if time_partition.is_none() {
259258
return updated_schema;
260259
}
261-
let time_partition_field_name = time_partition.unwrap();
260+
262261
let new_schema: Vec<Field> = updated_schema
263262
.fields()
264263
.iter()
265264
.map(|field| {
266-
if *field.name() == time_partition_field_name {
265+
if field.name() == time_partition.unwrap() {
267266
if field.data_type() == &DataType::Utf8 {
268267
let new_data_type = DataType::Timestamp(TimeUnit::Millisecond, None);
269268
Field::new(field.name().clone(), new_data_type, true)

src/handlers/http/ingest.rs

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -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,

src/handlers/http/modal/utils/ingest_utils.rs

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -230,7 +230,11 @@ pub fn into_event_batch(
230230
tags,
231231
metadata,
232232
};
233-
let (rb, is_first) = event.into_recordbatch(schema, static_schema_flag, time_partition)?;
233+
let (rb, is_first) = event.into_recordbatch(
234+
&schema,
235+
static_schema_flag.as_ref(),
236+
time_partition.as_ref(),
237+
)?;
234238
Ok((rb, is_first))
235239
}
236240

0 commit comments

Comments
 (0)