From d7043ac91c7d831dfb34fcff9815a25886b3cc8a Mon Sep 17 00:00:00 2001 From: Ashish Kolhe <35160958+oasisk@users.noreply.github.com> Date: Thu, 13 Apr 2023 19:55:45 +0530 Subject: [PATCH] fix: function update & disble udf (#564) --- README.md | 2 +- src/handler/http/request/functions/mod.rs | 2 - src/handler/http/router/mod.rs | 1 + src/meta/functions.rs | 2 +- src/service/functions.rs | 3 + src/service/search/datafusion/exec.rs | 6 +- .../search/datafusion/transform_udf.rs | 23 ++-- web/src/components/logstream/schema.vue | 113 +++++------------- 8 files changed, 52 insertions(+), 100 deletions(-) diff --git a/README.md b/README.md index 9dea4bf1be..b343a0dc13 100644 --- a/README.md +++ b/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. diff --git a/src/handler/http/request/functions/mod.rs b/src/handler/http/request/functions/mod.rs index 2b2b5251fb..bbde888789 100644 --- a/src/handler/http/request/functions/mod.rs +++ b/src/handler/http/request/functions/mod.rs @@ -109,7 +109,6 @@ async fn delete_function(path: web::Path<(String, String)>) -> Result) -> Result 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)] diff --git a/src/service/functions.rs b/src/service/functions.rs index 971df14107..64b3a6b679 100644 --- a/src/service/functions.rs +++ b/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 { diff --git a/src/service/search/datafusion/exec.rs b/src/service/search/datafusion/exec.rs index 678afc11b8..48e377d235 100644 --- a/src/service/search/datafusion/exec.rs +++ b/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)] diff --git a/src/service/search/datafusion/transform_udf.rs b/src/service/search/datafusion/transform_udf.rs index 75f52742fb..0ce8334192 100644 --- a/src/service/search/datafusion/transform_udf.rs +++ b/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 { +pub async fn _get_all_transform(org_id: &str) -> Vec { 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 Option { +pub fn _compile_vrl_function(func: &str) -> Option { let result = vrl::compile(func, &vrl_stdlib::all()); match result { Ok(CompilationResult { @@ -135,7 +135,7 @@ pub fn compile_vrl_function(func: &str) -> Option { } } -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", diff --git a/web/src/components/logstream/schema.vue b/web/src/components/logstream/schema.vue index e41998b173..79712227df 100644 --- a/web/src/components/logstream/schema.vue +++ b/web/src/components/logstream/schema.vue @@ -18,31 +18,19 @@
-
+
{{ t("logStream.schemaHeader") }}
- +
-
+
No data available.
@@ -98,85 +86,50 @@
{{ t("logStream.mapping") }} - +
-
- +
+
- - - - - - - - + + + + + + + + - +
{{ t("logStream.propertyName") }}{{ t("logStream.propertyType") }}{{ t("logStream.streamftsKey") }}{{ t("logStream.streamPartitionKey") }}
{{ t("logStream.propertyName") }}{{ t("logStream.propertyType") }}{{ t("logStream.streamftsKey") }}{{ t("logStream.streamPartitionKey") }}
{{ schema.name }} {{ schema.type }} - - - - {{ schema.level }} +
-
- - +
+ +
@@ -346,7 +299,7 @@ export default defineComponent({ onSubmit, updateSettingsForm, format, - getImageURL: () => {}, + getImageURL: () => { }, }; }, created() {