remove redundancy

Author: nikhilsinhaparseable

src/handlers/http/ingest.rs

@@ -118,18 +118,12 @@ pub async fn ingest(
 
     //if stream exists, fetch the stream log source
     //return error if the stream log source is otel traces or otel metrics
-    let stream = validate_stream_for_ingestion(&stream_name)?;
+    validate_stream_for_ingestion(&stream_name)?;
 
     PARSEABLE
         .add_update_log_source(&stream_name, log_source_entry)
         .await?;
 
-    if stream.get_time_partition().is_some() {
-        return Err(PostError::CustomError(
-            "Ingestion is not allowed to stream with time partition".to_string(),
-        ));
-    }
-
     flatten_and_push_logs(json, &stream_name, &log_source, &p_custom_fields, None).await?;
 
     Ok(HttpResponse::Ok().finish())
@@ -397,13 +391,7 @@ pub async fn post_event(
 
     //if stream exists, fetch the stream log source
     //return error if the stream log source is otel traces or otel metrics
-    let stream = validate_stream_for_ingestion(&stream_name)?;
-
-    if stream.get_time_partition().is_some() {
-        return Err(PostError::Invalid(anyhow::anyhow!(
-            "Ingestion is not allowed to stream with time partition. Please upgrade to Parseable enterprise to enable this feature"
-        )));
-    }
+    validate_stream_for_ingestion(&stream_name)?;
 
     flatten_and_push_logs(json, &stream_name, &log_source, &p_custom_fields, None).await?;
 

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

@@ -23,7 +23,7 @@ use opentelemetry_proto::tonic::{
     logs::v1::LogsData, metrics::v1::MetricsData, trace::v1::TracesData,
 };
 use serde_json::Value;
-use std::{collections::HashMap, sync::Arc};
+use std::collections::HashMap;
 use tracing::warn;
 
 use crate::{
@@ -39,7 +39,7 @@ use crate::{
         },
     },
     otel::{logs::flatten_otel_logs, metrics::flatten_otel_metrics, traces::flatten_otel_traces},
-    parseable::{PARSEABLE, Stream},
+    parseable::PARSEABLE,
     storage::StreamType,
     utils::json::{convert_array_to_object, flatten::convert_to_array},
 };
@@ -268,7 +268,7 @@ fn verify_dataset_fields_count(stream_name: &str) -> Result<(), PostError> {
     Ok(())
 }
 
-pub fn validate_stream_for_ingestion(stream_name: &str) -> Result<Arc<Stream>, PostError> {
+pub fn validate_stream_for_ingestion(stream_name: &str) -> Result<(), PostError> {
     let stream = PARSEABLE.get_stream(stream_name)?;
 
     // Validate that the stream's log source is compatible
@@ -283,12 +283,12 @@ pub fn validate_stream_for_ingestion(stream_name: &str) -> Result<Arc<Stream>, P
 
     // Check for time partition
     if stream.get_time_partition().is_some() {
-        return Err(PostError::CustomError(
-            "Ingestion is not allowed to stream with time partition".to_string(),
-        ));
+        return Err(PostError::Invalid(anyhow::anyhow!(
+            "Ingestion with time partition is not supported in Parseable OSS"
+        )));
     }
 
-    Ok(stream)
+    Ok(())
 }
 
 #[cfg(test)]