fix: function update & disble udf (#564)

README.md

@@ -1,4 +1,4 @@
-🚀 10x easier, 🚀 10x cheaper, 🚀 petabyte scale - Elasticsearch/Splunk/Datadog alternative for 🚀 (logs, metrics, traces).
+🚀 10x easier, 🚀 100x lower storage cost, 🚀 high performance, 🚀 petabyte scale - Elasticsearch/Splunk/Datadog alternative for 🚀 (logs, metrics, traces).
 
 # ZincObserve
 ZincObserve is a cloud native observability platform built specifically for logs, metrics, traces and analytics designed to work at petabyte scale.

src/handler/http/request/functions/mod.rs

@@ -109,7 +109,6 @@ async fn delete_function(path: web::Path<(String, String)>) -> Result<HttpRespon
     ),
     params(
         ("org_id" = String, Path, description = "Organization name"),
-        ("stream_name" = String, Path, description = "Stream name"),
         ("name" = String, Path, description = "Function name"),
     ),
     request_body(content = Transform, description = "Function data", content_type = "application/json"),
@@ -352,7 +351,6 @@ async fn delete_function(_path: web::Path<(String, String)>) -> Result<HttpRespo
     ),
     params(
         ("org_id" = String, Path, description = "Organization name"),
-        ("stream_name" = String, Path, description = "Stream name"),
         ("name" = String, Path, description = "Function name"),
     ),
     request_body(content = Transform, description = "Function data", content_type = "application/json"),

src/handler/http/router/mod.rs

@@ -131,6 +131,7 @@ pub fn get_service_routes(cfg: &mut web::ServiceConfig) {
             .service(functions::save_function)
             .service(functions::list_functions)
             .service(functions::delete_function)
+            .service(functions::update_function)
             .service(functions::add_function_to_stream)
             .service(functions::list_stream_functions)
             .service(functions::delete_stream_function)

src/meta/functions.rs

@@ -86,7 +86,7 @@ impl Transform {
 
 impl PartialEq for Transform {
     fn eq(&self, other: &Self) -> bool {
-        self.name == other.name && self.function == other.function
+        self.name == other.name && self.function == other.function && self.params == other.params
     }
 }
 #[derive(Clone, Debug, Serialize, Deserialize, ToSchema)]

src/service/functions.rs

@@ -80,6 +80,9 @@ pub async fn update_function(
         return Ok(HttpResponse::Ok().json(func));
     }
 
+    // UI mostly like in 1st version wont send streams, so we need to add them back from existing function
+    func.streams = existing_fn.streams;
+
     extract_num_args(&mut func);
     let name = func.name.to_owned();
     if let Err(error) = db::functions::set(org_id.as_str(), name.as_str(), func).await {

src/service/search/datafusion/exec.rs

@@ -38,8 +38,6 @@ use std::time::Instant;
 use uuid::Uuid;
 
 use super::storage::file_list;
-#[cfg(feature = "zo_functions")]
-use super::transform_udf::get_all_transform;
 use crate::common::json;
 use crate::infra::cache::tmpfs;
 use crate::infra::config::{get_parquet_compression, CONFIG};
@@ -784,13 +782,13 @@ async fn register_udf(ctx: &mut SessionContext, _org_id: &str) {
     ctx.register_udf(super::regexp_udf::REGEX_MATCH_UDF.clone());
     ctx.register_udf(super::regexp_udf::REGEX_NOT_MATCH_UDF.clone());
     ctx.register_udf(super::time_range_udf::TIME_RANGE_UDF.clone());
-    #[cfg(feature = "zo_functions")]
+    /*     #[cfg(feature = "zo_functions")]
     {
         let udf_list = get_all_transform(_org_id).await;
         for udf in udf_list {
             ctx.register_udf(udf.clone());
         }
-    }
+    } */
 }
 
 #[cfg(test)]

src/service/search/datafusion/transform_udf.rs

@@ -27,7 +27,7 @@ use vrl::{CompilationResult, Program, Runtime};
 
 use crate::infra::config::QUERY_FUNCTIONS;
 
-fn create_user_df(
+fn _create_user_df(
     fn_name: &str,
     num_args: u8,
     pow_scalar: ScalarFunctionImplementation,
@@ -46,13 +46,13 @@ fn create_user_df(
     )
 }
 
-pub async fn get_all_transform(org_id: &str) -> Vec<datafusion::logical_expr::ScalarUDF> {
+pub async fn _get_all_transform(org_id: &str) -> Vec<datafusion::logical_expr::ScalarUDF> {
     let mut udf;
     let mut udf_list = Vec::new();
     for transform in QUERY_FUNCTIONS.iter() {
         let key = transform.key();
         if key.contains(org_id) {
-            udf = get_udf_vrl(
+            udf = _get_udf_vrl(
                 transform.name.to_owned(),
                 transform.function.to_owned().as_str(),
                 &transform.params,
@@ -65,7 +65,7 @@ pub async fn get_all_transform(org_id: &str) -> Vec<datafusion::logical_expr::Sc
     udf_list
 }
 
-fn get_udf_vrl(
+fn _get_udf_vrl(
     fn_name: String,
     func: &str,
     params: &str,
@@ -96,8 +96,8 @@ fn get_udf_vrl(
                 ));
             }
             obj_str.push_str(&format!(" \n {}", &local_func));
-            if let Some(func) = compile_vrl_function(&obj_str) {
-                data_vec.insert(i, apply_vrl_fn(&mut runtime, func));
+            if let Some(func) = _compile_vrl_function(&obj_str) {
+                data_vec.insert(i, _apply_vrl_fn(&mut runtime, func));
             } else {
                 data_vec.insert(i, "".to_string());
             }
@@ -116,11 +116,11 @@ fn get_udf_vrl(
     // * give it a name so that it shows nicely when the plan is printed
     // * declare what input it expects
     // * declare its return type
-    let pow_udf = create_user_df(local_fn_name.as_str(), num_args, pow_scalar);
+    let pow_udf = _create_user_df(local_fn_name.as_str(), num_args, pow_scalar);
     pow_udf
 }
 
-pub fn compile_vrl_function(func: &str) -> Option<Program> {
+pub fn _compile_vrl_function(func: &str) -> Option<Program> {
     let result = vrl::compile(func, &vrl_stdlib::all());
     match result {
         Ok(CompilationResult {
@@ -135,7 +135,7 @@ pub fn compile_vrl_function(func: &str) -> Option<Program> {
     }
 }
 
-pub fn apply_vrl_fn(runtime: &mut Runtime, program: vrl::Program) -> String {
+pub fn _apply_vrl_fn(runtime: &mut Runtime, program: vrl::Program) -> String {
     let obj_str = String::from("");
 
     let mut metadata = vrl_value::Value::from(BTreeMap::new());
@@ -159,8 +159,7 @@ pub fn apply_vrl_fn(runtime: &mut Runtime, program: vrl::Program) -> String {
 
 #[cfg(test)]
 mod tests {
-    use super::*;
-
+    use crate::service::search::datafusion::transform_udf::_get_udf_vrl;
     use datafusion::arrow::array::{Int64Array, StringArray};
     use datafusion::arrow::datatypes::{DataType, Field, Schema};
     use datafusion::arrow::record_batch::RecordBatch;
@@ -193,7 +192,7 @@ mod tests {
         // declare a new context. In spark API, this corresponds to a new spark SQLsession
         let ctx = SessionContext::new();
 
-        let vrl_udf = get_udf_vrl(
+        let vrl_udf = _get_udf_vrl(
             "vrlconcat".to_string(),
             ". = col1  + col2 \n .",
             "col1,col2",

web/src/components/logstream/schema.vue

@@ -18,31 +18,19 @@
     <q-card-section class="q-pa-md">
       <div class="row items-center no-wrap">
         <div class="col">
-          <div
-            class="text-body1 text-bold text-dark"
-            data-test="schema-title-text"
-          >
+          <div class="text-body1 text-bold text-dark" data-test="schema-title-text">
             {{ t("logStream.schemaHeader") }}
           </div>
         </div>
         <div class="col-auto">
-          <q-btn
-            v-close-popup
-            round
-            flat
-            :icon="'img:' + getImageURL('images/common/close_icon.svg')"
-          />
+          <q-btn v-close-popup round flat :icon="'img:' + getImageURL('images/common/close_icon.svg')" />
         </div>
       </div>
     </q-card-section>
     <q-separator />
     <q-card-section>
       <q-form ref="updateSettingsForm" @submit.prevent="onSubmit">
-        <div
-          v-if="indexData.schema.length == 0"
-          class="q-pt-md text-center q-w-md q-mx-lg"
-          style="max-width: 450px"
-        >
+        <div v-if="indexData.schema.length == 0" class="q-pt-md text-center q-w-md q-mx-lg" style="max-width: 450px">
           No data available.
         </div>
         <div v-else class="indexDetailsContainer">
@@ -98,85 +86,50 @@
 
           <div class="title" data-test="schema-log-stream-mapping-title-text">
             {{ t("logStream.mapping") }}
-            <label
-              v-show="indexData.defaultFts"
-              class="warning-msg"
-              style="font-weight: normal"
-              >- Using default fts keys, as no fts keys are set for
-              stream.</label
-            >
+            <label v-show="indexData.defaultFts" class="warning-msg" style="font-weight: normal">- Using default fts keys,
+              as no fts keys are set for
+              stream.</label>
           </div>
 
           <!-- Note: Drawer max-height to be dynamically calculated with JS -->
-          <div
-            class="q-table__container q-table--cell-separator"
-            style="height: calc(100vh - 460px); overflow: auto"
-          >
-            <table
-              class="q-table"
-              data-test="schema-log-stream-field-mapping-table"
-            >
+          <div class="q-table__container q-table--cell-separator" style="height: calc(100vh - 460px); overflow: auto">
+            <table class="q-table" data-test="schema-log-stream-field-mapping-table">
               <thead>
                 <tr>
-                  <th>{{ t("logStream.propertyName") }}</th>
-                  <th>{{ t("logStream.propertyType") }}</th>
-                  <th>{{ t("logStream.streamftsKey") }}</th>
-                  <th>{{ t("logStream.streamPartitionKey") }}</th>
-                </tr>
-              </thead>
-              <tbody>
-                <tr
-                  v-for="(schema, index) in indexData.schema"
-                  :key="index + '_' + schema.name"
-                  class="list-item"
-                >
+                <th>{{ t("logStream.propertyName") }}</th>
+                <th>{{ t("logStream.propertyType") }}</th>
+                <th>{{ t("logStream.streamftsKey") }}</th>
+                <th>{{ t("logStream.streamPartitionKey") }}</th>
+              </tr>
+            </thead>
+            <tbody>
+              <tr v-for="(schema, index) in indexData.schema" :key="index + '_' + schema.name" class="list-item">
                   <td>{{ schema.name }}</td>
                   <td>{{ schema.type }}</td>
                   <td class="text-center">
-                    <q-checkbox
-                      :data-test="`schema-stream-${schema.name}-field-fts-key-checkbox`"
-                      v-model="schema.ftsKey"
-                      size="sm"
-                    />
-                  </td>
-                  <td class="text-center">
-                    <q-checkbox
-                      :data-test="`schema-stream-${schema.name}-field-partition-key-checkbox`"
-                      v-model="schema.partitionKey"
-                      size="sm"
-                    >
-                      {{ schema.level }}</q-checkbox
-                    >
+                    <q-checkbox :data-test="`schema-stream-${schema.name}-field-fts-key-checkbox`" v-model="schema.ftsKey"
+                      size="sm" />
                   </td>
+                  <!--<td class="text-center">
+                      <q-checkbox
+                        :data-test="`schema-stream-${schema.name}-field-partition-key-checkbox`"
+                        v-model="schema.partitionKey"
+                        size="sm"
+                      >
+                        {{ schema.level }}</q-checkbox
+                      >
+                    </td>-->
                 </tr>
               </tbody>
             </table>
           </div>
         </div>
 
-        <div
-          v-if="indexData.schema.length > 0"
-          class="flex justify-center q-mt-sm"
-        >
-          <q-btn
-            v-close-popup
-            data-test="schema-cancel-button"
-            class="q-mb-md text-bold no-border"
-            :label="t('logStream.cancel')"
-            text-color="light-text"
-            padding="sm md"
-            color="accent"
-            no-caps
-          />
-          <q-btn
-            data-test="schema-update-settings-button"
-            :label="t('logStream.updateSettings')"
-            class="q-mb-md text-bold no-border q-ml-md"
-            color="secondary"
-            padding="sm xl"
-            type="submit"
-            no-caps
-          />
+        <div v-if="indexData.schema.length > 0" class="flex justify-center q-mt-sm">
+          <q-btn v-close-popup data-test="schema-cancel-button" class="q-mb-md text-bold no-border"
+            :label="t('logStream.cancel')" text-color="light-text" padding="sm md" color="accent" no-caps />
+          <q-btn data-test="schema-update-settings-button" :label="t('logStream.updateSettings')"
+            class="q-mb-md text-bold no-border q-ml-md" color="secondary" padding="sm xl" type="submit" no-caps />
         </div>
       </q-form>
     </q-card-section>
@@ -346,7 +299,7 @@ export default defineComponent({
       onSubmit,
       updateSettingsForm,
       format,
-      getImageURL: () => {},
+      getImageURL: () => { },
     };
   },
   created() {