Merge Changes from stratchpad

Author: Eshanatnight

server/src/handlers/http.rs

@@ -59,6 +59,7 @@ pub fn base_path_without_preceding_slash() -> String {
     base_path().trim_start_matches('/').to_string()
 }
 
+
 pub async fn send_schema_request(stream_name: &str) -> anyhow::Result<Vec<arrow_schema::Schema>> {
     let mut res = vec![];
     let ima = QueryServer::get_ingester_info().await.unwrap();

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

@@ -175,6 +175,14 @@ impl IngestServer {
                             )
                             .app_data(web::PayloadConfig::default().limit(MAX_EVENT_PAYLOAD_SIZE)),
                     )
+                    .service(
+                        // GET "/logstream/{logstream}/schema" ==> Get schema for given log stream
+                        web::resource("/schema").route(
+                            web::get()
+                                .to(logstream::schema)
+                                .authorize_for_stream(Action::GetSchema),
+                        ),
+                    )
                     .service(
                         // GET "/logstream/{logstream}/stats" ==> Get stats for given log stream
                         web::resource("/stats").route(

server/src/handlers/http/query.rs

@@ -19,7 +19,7 @@
 use actix_web::http::header::ContentType;
 use actix_web::web::{self, Json};
 use actix_web::{FromRequest, HttpRequest, Responder};
-use arrow_schema::Schema;
+use arrow_json::reader::infer_json_schema_from_iterator;
 use chrono::{DateTime, Utc};
 use datafusion::error::DataFusionError;
 use datafusion::execution::context::SessionState;
@@ -30,8 +30,13 @@ use std::pin::Pin;
 use std::sync::Arc;
 use std::time::Instant;
 
-use crate::event::commit_schema;
+// Eshan's Code Under test
+#[allow(unused_imports)]
+use arrow_schema::Schema;
+#[allow(unused_imports)]
 use crate::handlers::http::send_schema_request;
+
+use crate::event::commit_schema;
 use crate::metrics::QUERY_EXECUTE_TIME;
 use crate::option::{Mode, CONFIG};
 use crate::query::error::ExecuteError;
@@ -63,28 +68,41 @@ pub async fn query(req: HttpRequest, query_request: Query) -> Result<impl Respon
     let session_state = QUERY_SESSION.state();
     let mut query = into_query(&query_request, &session_state).await?;
 
-    if CONFIG.parseable.mode == Mode::Query {
-        if let Ok(schs) = send_schema_request(&query.table_name().unwrap()).await {
-            let new_schema =
-                Schema::try_merge(schs).map_err(|err| QueryError::Custom(err.to_string()))?;
+    // if CONFIG.parseable.mode == Mode::Query {
+    //     if let Ok(schs) = send_schema_request(&query.table_name().unwrap()).await {
+    //         let new_schema =
+    //             Schema::try_merge(schs).map_err(|err| QueryError::Custom(err.to_string()))?;
 
-            commit_schema(&query.table_name().unwrap(), Arc::new(new_schema.clone()))
-                .map_err(|err| QueryError::Custom(format!("Error committing schema: {}", err)))?;
+    //         commit_schema(&query.table_name().unwrap(), Arc::new(new_schema.clone()))
+    //             .map_err(|err| QueryError::Custom(format!("Error committing schema: {}", err)))?;
 
-            commit_schema_to_storage(&query.table_name().unwrap(), new_schema)
-                .await
-                .map_err(|err| {
-                    QueryError::Custom(format!("Error committing schema to storage\nError:{err}"))
-                })?;
-        }
-    }
+    //         commit_schema_to_storage(&query.table_name().unwrap(), new_schema)
+    //             .await
+    //             .map_err(|err| {
+    //                 QueryError::Custom(format!("Error committing schema to storage\nError:{err}"))
+    //             })?;
+    //     }
+    // }
 
     let mmem = if CONFIG.parseable.mode == Mode::Query {
         // create a new query to send to the ingestors
         if let Some(que) = transform_query_for_ingestor(&query_request) {
             let vals = send_request_to_ingestor(&que)
                 .await
                 .map_err(|err| QueryError::Custom(err.to_string()))?;
+            let infered_schema = infer_json_schema_from_iterator(vals.iter().map(Ok)).map_err(|err| {
+                QueryError::Custom(format!("Error inferring schema from iterator\nError:{err}"))
+            })?;
+
+            commit_schema(&query.table_name().unwrap(), Arc::new(infered_schema.clone()))
+                .map_err(|err| QueryError::Custom(format!("Error committing schema: {}", err)))?;
+
+            commit_schema_to_storage(&query.table_name().unwrap(), infered_schema)
+                .await
+                .map_err(|err| {
+                    QueryError::Custom(format!("Error committing schema to storage\nError:{err}"))
+                })?;
+
             Some(vals)
         } else {
             None

server/src/query.rs

@@ -51,6 +51,7 @@ pub static QUERY_SESSION: Lazy<SessionContext> =
     Lazy::new(|| Query::create_session_context(CONFIG.storage()));
 
 // A query request by client
+#[derive(Debug)]
 pub struct Query {
     pub raw_logical_plan: LogicalPlan,
     pub start: DateTime<Utc>,
@@ -102,10 +103,7 @@ impl Query {
         SessionContext::new_with_state(state)
     }
 
-    pub async fn execute(
-        &self,
-        mem: Option<Vec<Value>>,
-    ) -> Result<(Vec<RecordBatch>, Vec<String>), ExecuteError> {
+    pub async fn execute(&self) -> Result<(Vec<RecordBatch>, Vec<String>), ExecuteError> {
         let df = QUERY_SESSION
             .execute_logical_plan(self.final_logical_plan())
             .await?;

server/src/response.rs

@@ -30,7 +30,7 @@ pub struct QueryResponse {
 }
 
 impl QueryResponse {
-    pub fn to_http(&self) -> impl Responder {
+    pub fn to_http(&self, imem: Option<Vec<Value>>) -> impl Responder {
         log::info!("{}", "Returning query results");
         let records: Vec<&RecordBatch> = self.records.iter().collect();
         let mut json_records = record_batches_to_json_rows(&records).unwrap();