pipeline/
etl.rs

1// Copyright 2023 Greptime Team
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15#![allow(dead_code)]
16pub mod ctx_req;
17pub mod field;
18pub mod processor;
19pub mod transform;
20pub mod value;
21
22use std::collections::HashMap;
23
24use api::v1::Row;
25use common_time::timestamp::TimeUnit;
26use itertools::Itertools;
27use processor::{Processor, Processors};
28use snafu::{OptionExt, ResultExt, ensure};
29use transform::Transforms;
30use vrl::core::Value as VrlValue;
31use yaml_rust::{Yaml, YamlLoader};
32
33use crate::dispatcher::{Dispatcher, Rule};
34use crate::error::{
35    ArrayElementMustBeObjectSnafu, AutoTransformOneTimestampSnafu, Error,
36    IntermediateKeyIndexSnafu, InvalidVersionNumberSnafu, Result, TransformArrayElementSnafu,
37    YamlLoadSnafu, YamlParseSnafu,
38};
39use crate::etl::processor::ProcessorKind;
40use crate::etl::transform::transformer::greptime::{
41    RowWithTableSuffix, values_to_row, values_to_rows,
42};
43use crate::tablesuffix::TableSuffixTemplate;
44use crate::{
45    ContextOpt, GreptimeTransformer, IdentityTimeIndex, PipelineContext, SchemaInfo,
46    unwrap_or_continue_if_err,
47};
48
49const DESCRIPTION: &str = "description";
50const DOC_VERSION: &str = "version";
51const PROCESSORS: &str = "processors";
52const TRANSFORM: &str = "transform";
53const TRANSFORMS: &str = "transforms";
54const DISPATCHER: &str = "dispatcher";
55const TABLESUFFIX: &str = "table_suffix";
56
57pub enum Content<'a> {
58    Json(&'a str),
59    Yaml(&'a str),
60}
61
62pub fn parse(input: &Content) -> Result<Pipeline> {
63    match input {
64        Content::Yaml(str) => {
65            let docs = YamlLoader::load_from_str(str).context(YamlLoadSnafu)?;
66
67            ensure!(docs.len() == 1, YamlParseSnafu);
68
69            let doc = &docs[0];
70
71            let description = doc[DESCRIPTION].as_str().map(|s| s.to_string());
72
73            let doc_version = (&doc[DOC_VERSION]).try_into()?;
74
75            let processors = if let Some(v) = doc[PROCESSORS].as_vec() {
76                v.try_into()?
77            } else {
78                Processors::default()
79            };
80
81            let transformers = if let Some(v) = doc[TRANSFORMS].as_vec().or(doc[TRANSFORM].as_vec())
82            {
83                v.try_into()?
84            } else {
85                Transforms::default()
86            };
87
88            let transformer = if transformers.is_empty() {
89                // use auto transform
90                // check processors have at least one timestamp-related processor
91                let cnt = processors
92                    .iter()
93                    .filter_map(|p| match p {
94                        ProcessorKind::Date(d) if !d.ignore_missing() => Some(
95                            d.fields
96                                .iter()
97                                .map(|f| (f.target_or_input_field(), TimeUnit::Nanosecond))
98                                .collect_vec(),
99                        ),
100                        ProcessorKind::Epoch(e) if !e.ignore_missing() => Some(
101                            e.fields
102                                .iter()
103                                .map(|f| (f.target_or_input_field(), (&e.resolution).into()))
104                                .collect_vec(),
105                        ),
106                        _ => None,
107                    })
108                    .flatten()
109                    .collect_vec();
110                ensure!(cnt.len() == 1, AutoTransformOneTimestampSnafu);
111
112                let (ts_name, timeunit) = cnt.first().unwrap();
113                TransformerMode::AutoTransform(ts_name.to_string(), *timeunit)
114            } else {
115                TransformerMode::GreptimeTransformer(GreptimeTransformer::new(
116                    transformers,
117                    &doc_version,
118                )?)
119            };
120
121            let dispatcher = if !doc[DISPATCHER].is_badvalue() {
122                Some(Dispatcher::try_from(&doc[DISPATCHER])?)
123            } else {
124                None
125            };
126
127            let tablesuffix = if !doc[TABLESUFFIX].is_badvalue() {
128                Some(TableSuffixTemplate::try_from(&doc[TABLESUFFIX])?)
129            } else {
130                None
131            };
132
133            Ok(Pipeline {
134                doc_version,
135                description,
136                processors,
137                transformer,
138                dispatcher,
139                tablesuffix,
140            })
141        }
142        Content::Json(_) => unimplemented!(),
143    }
144}
145
146#[derive(Debug, Default, Copy, Clone, PartialEq, Eq)]
147pub enum PipelineDocVersion {
148    /// 1. All fields meant to be preserved have to explicitly set in the transform section.
149    /// 2. Or no transform is set, then the auto-transform will be used.
150    #[default]
151    V1,
152
153    /// A combination of transform and auto-transform.
154    /// First it goes through the transform section,
155    /// then use auto-transform to set the rest fields.
156    ///
157    /// This is useful if you only want to set the index field,
158    /// and let the normal fields be auto-inferred.
159    V2,
160}
161
162impl TryFrom<&Yaml> for PipelineDocVersion {
163    type Error = Error;
164
165    fn try_from(value: &Yaml) -> Result<Self> {
166        if value.is_badvalue() || value.is_null() {
167            return Ok(PipelineDocVersion::V1);
168        }
169
170        let version = match value {
171            Yaml::String(s) => s
172                .parse::<i64>()
173                .map_err(|_| InvalidVersionNumberSnafu { version: s.clone() }.build())?,
174            Yaml::Integer(i) => *i,
175            _ => {
176                return InvalidVersionNumberSnafu {
177                    version: value.as_str().unwrap_or_default().to_string(),
178                }
179                .fail();
180            }
181        };
182
183        match version {
184            1 => Ok(PipelineDocVersion::V1),
185            2 => Ok(PipelineDocVersion::V2),
186            _ => InvalidVersionNumberSnafu {
187                version: version.to_string(),
188            }
189            .fail(),
190        }
191    }
192}
193
194#[derive(Debug)]
195pub struct Pipeline {
196    doc_version: PipelineDocVersion,
197    description: Option<String>,
198    processors: processor::Processors,
199    dispatcher: Option<Dispatcher>,
200    transformer: TransformerMode,
201    tablesuffix: Option<TableSuffixTemplate>,
202}
203
204#[derive(Debug, Clone)]
205pub enum TransformerMode {
206    GreptimeTransformer(GreptimeTransformer),
207    AutoTransform(String, TimeUnit),
208}
209
210/// Where the pipeline executed is dispatched to, with context information
211#[derive(Debug, Hash, PartialEq, Eq, Clone, PartialOrd, Ord)]
212pub struct DispatchedTo {
213    pub table_suffix: String,
214    pub pipeline: Option<String>,
215}
216
217impl From<&Rule> for DispatchedTo {
218    fn from(value: &Rule) -> Self {
219        DispatchedTo {
220            table_suffix: value.table_suffix.clone(),
221            pipeline: value.pipeline.clone(),
222        }
223    }
224}
225
226impl DispatchedTo {
227    /// Generate destination table name from input
228    pub fn dispatched_to_table_name(&self, original: &str) -> String {
229        [original, &self.table_suffix].concat()
230    }
231}
232
233/// The result of pipeline execution
234#[derive(Debug)]
235pub enum PipelineExecOutput {
236    Transformed(TransformedOutput),
237    DispatchedTo(DispatchedTo, VrlValue),
238    Filtered,
239}
240
241/// The result after processors and dispatcher rules have run.
242#[derive(Debug)]
243pub enum PipelineProcessOutput {
244    Processed(VrlValue),
245    DispatchedTo(DispatchedTo, VrlValue),
246    Filtered,
247}
248
249/// Output from a successful pipeline transformation.
250///
251/// Rows are grouped by their ContextOpt, with each row having its own optional
252/// table_suffix for routing to different tables when using one-to-many expansion.
253/// This enables true per-row configuration options where different rows can have
254/// different database settings (TTL, merge mode, etc.).
255#[derive(Debug)]
256pub struct TransformedOutput {
257    /// Rows grouped by their ContextOpt, each with optional table suffix
258    pub rows_by_context: HashMap<ContextOpt, Vec<RowWithTableSuffix>>,
259}
260
261impl PipelineExecOutput {
262    // Note: This is a test only function, do not use it in production.
263    pub fn into_transformed(self) -> Option<Vec<RowWithTableSuffix>> {
264        if let Self::Transformed(TransformedOutput { rows_by_context }) = self {
265            // For backward compatibility, merge all rows with a default ContextOpt
266            Some(rows_by_context.into_values().flatten().collect())
267        } else {
268            None
269        }
270    }
271
272    // New method for accessing the HashMap structure directly
273    pub fn into_transformed_hashmap(self) -> Option<HashMap<ContextOpt, Vec<RowWithTableSuffix>>> {
274        if let Self::Transformed(TransformedOutput { rows_by_context }) = self {
275            Some(rows_by_context)
276        } else {
277            None
278        }
279    }
280
281    // Backward compatibility helper that returns first ContextOpt with all its rows
282    // or merges all rows with default ContextOpt for multi-context scenarios
283    pub fn into_legacy_format(self) -> Option<(ContextOpt, Vec<RowWithTableSuffix>)> {
284        if let Self::Transformed(TransformedOutput { rows_by_context }) = self {
285            if rows_by_context.len() == 1 {
286                let (opt, rows) = rows_by_context.into_iter().next().unwrap();
287                Some((opt, rows))
288            } else {
289                // Multiple contexts: merge all rows with default ContextOpt for test compatibility
290                let all_rows: Vec<RowWithTableSuffix> =
291                    rows_by_context.into_values().flatten().collect();
292                Some((ContextOpt::default(), all_rows))
293            }
294        } else {
295            None
296        }
297    }
298
299    // Note: This is a test only function, do not use it in production.
300    pub fn into_dispatched(self) -> Option<DispatchedTo> {
301        if let Self::DispatchedTo(d, _) = self {
302            Some(d)
303        } else {
304            None
305        }
306    }
307}
308
309impl Pipeline {
310    fn is_v1(&self) -> bool {
311        self.doc_version == PipelineDocVersion::V1
312    }
313
314    pub fn exec_mut(
315        &self,
316        val: VrlValue,
317        pipeline_ctx: &PipelineContext<'_>,
318        schema_info: &mut SchemaInfo,
319    ) -> Result<PipelineExecOutput> {
320        match self.process_mut(val)? {
321            PipelineProcessOutput::Processed(val) => self
322                .transform_mut(val, pipeline_ctx, schema_info)
323                .map(PipelineExecOutput::Transformed),
324            PipelineProcessOutput::DispatchedTo(dispatched_to, val) => {
325                Ok(PipelineExecOutput::DispatchedTo(dispatched_to, val))
326            }
327            PipelineProcessOutput::Filtered => Ok(PipelineExecOutput::Filtered),
328        }
329    }
330
331    pub fn process_mut(&self, mut val: VrlValue) -> Result<PipelineProcessOutput> {
332        for processor in self.processors.iter() {
333            val = processor.exec_mut(val)?;
334            if val.is_null() {
335                return Ok(PipelineProcessOutput::Filtered);
336            }
337        }
338
339        if let Some(rule) = self.dispatcher.as_ref().and_then(|d| d.exec(&val)) {
340            return Ok(PipelineProcessOutput::DispatchedTo(rule.into(), val));
341        }
342
343        Ok(PipelineProcessOutput::Processed(val))
344    }
345
346    pub fn transform_mut(
347        &self,
348        val: VrlValue,
349        pipeline_ctx: &PipelineContext<'_>,
350        schema_info: &mut SchemaInfo,
351    ) -> Result<TransformedOutput> {
352        let mut val = if val.is_array() {
353            val
354        } else {
355            VrlValue::Array(vec![val])
356        };
357
358        let rows_by_context = match self.transformer() {
359            TransformerMode::GreptimeTransformer(greptime_transformer) => {
360                transform_array_elements_by_ctx(
361                    // SAFETY: by line 326, val must be an array
362                    val.as_array_mut().unwrap(),
363                    greptime_transformer,
364                    self.is_v1(),
365                    schema_info,
366                    pipeline_ctx,
367                    self.tablesuffix.as_ref(),
368                )?
369            }
370            TransformerMode::AutoTransform(ts_name, time_unit) => {
371                let def = crate::PipelineDefinition::GreptimeIdentityPipeline(Some(
372                    IdentityTimeIndex::Epoch(ts_name.clone(), *time_unit, false),
373                ));
374                let n_ctx =
375                    PipelineContext::new(&def, pipeline_ctx.pipeline_param, pipeline_ctx.channel);
376                values_to_rows(
377                    schema_info,
378                    val,
379                    &n_ctx,
380                    None,
381                    true,
382                    self.tablesuffix.as_ref(),
383                )?
384            }
385        };
386
387        Ok(TransformedOutput { rows_by_context })
388    }
389
390    pub fn processors(&self) -> &processor::Processors {
391        &self.processors
392    }
393
394    pub fn transformer(&self) -> &TransformerMode {
395        &self.transformer
396    }
397
398    pub fn resolve_table_suffix(&self, value: &VrlValue) -> Option<String> {
399        ContextOpt::resolve_table_suffix(self.tablesuffix.as_ref(), value)
400    }
401
402    // the method is for test purpose
403    pub fn schemas(&self) -> Option<&Vec<greptime_proto::v1::ColumnSchema>> {
404        match &self.transformer {
405            TransformerMode::GreptimeTransformer(t) => Some(t.schemas()),
406            TransformerMode::AutoTransform(_, _) => None,
407        }
408    }
409
410    pub fn is_variant_table_name(&self) -> bool {
411        // even if the pipeline doesn't have dispatcher or table_suffix,
412        // it can still be a variant because of VRL processor and hint
413        self.dispatcher.is_some() || self.tablesuffix.is_some()
414    }
415}
416
417/// Transforms an array of VRL values into rows grouped by their ContextOpt.
418/// Each element can have its own ContextOpt for per-row configuration.
419fn transform_array_elements_by_ctx(
420    arr: &mut [VrlValue],
421    transformer: &GreptimeTransformer,
422    is_v1: bool,
423    schema_info: &mut SchemaInfo,
424    pipeline_ctx: &PipelineContext<'_>,
425    tablesuffix_template: Option<&TableSuffixTemplate>,
426) -> Result<HashMap<ContextOpt, Vec<RowWithTableSuffix>>> {
427    let skip_error = pipeline_ctx.pipeline_param.skip_error();
428    let mut rows_by_context = HashMap::new();
429
430    for (index, element) in arr.iter_mut().enumerate() {
431        if !element.is_object() {
432            unwrap_or_continue_if_err!(
433                ArrayElementMustBeObjectSnafu {
434                    index,
435                    actual_type: element.kind_str().to_string(),
436                }
437                .fail(),
438                skip_error
439            );
440        }
441
442        let table_suffix = ContextOpt::resolve_table_suffix(tablesuffix_template, element);
443        let values = unwrap_or_continue_if_err!(
444            transformer.transform_mut_with_schema(
445                element,
446                is_v1,
447                schema_info,
448                table_suffix.as_deref(),
449            ),
450            skip_error
451        );
452        if is_v1 {
453            // v1 mode: just use transformer output directly
454            let opt = unwrap_or_continue_if_err!(
455                ContextOpt::from_pipeline_map_to_opt(element),
456                skip_error
457            );
458            rows_by_context
459                .entry(opt)
460                .or_insert_with(Vec::new)
461                .push((Row { values }, table_suffix));
462        } else {
463            // v2 mode: combine with auto-transform for remaining fields
464            let mut value = element.clone();
465            let opt = unwrap_or_continue_if_err!(
466                ContextOpt::from_pipeline_map_to_opt(&mut value),
467                skip_error
468            );
469            let row = unwrap_or_continue_if_err!(
470                values_to_row(schema_info, value, pipeline_ctx, Some(values), false,)
471                    .map_err(Box::new)
472                    .context(TransformArrayElementSnafu { index }),
473                skip_error
474            );
475            rows_by_context
476                .entry(opt)
477                .or_default()
478                .push((row, table_suffix));
479        }
480    }
481
482    Ok(rows_by_context)
483}
484
485pub(crate) fn find_key_index(intermediate_keys: &[String], key: &str, kind: &str) -> Result<usize> {
486    intermediate_keys
487        .iter()
488        .position(|k| k == key)
489        .context(IntermediateKeyIndexSnafu { kind, key })
490}
491
492/// This macro is test only, do not use it in production.
493/// The schema_info cannot be used in auto-transform ts-infer mode for lacking the ts schema.
494///
495/// Usage:
496/// ```ignore
497/// let (pipeline, schema_info, pipeline_def, pipeline_param) = setup_pipeline!(pipeline);
498/// let pipeline_ctx = PipelineContext::new(&pipeline_def, &pipeline_param, Channel::Unknown);
499/// ```
500#[macro_export]
501macro_rules! setup_pipeline {
502    ($pipeline:expr) => {{
503        use std::sync::Arc;
504
505        use $crate::{GreptimePipelineParams, Pipeline, PipelineDefinition, SchemaInfo};
506
507        let pipeline: Arc<Pipeline> = Arc::new($pipeline);
508        let schema = pipeline.schemas().unwrap();
509        let schema_info = SchemaInfo::from_schema_list(schema.clone());
510
511        let pipeline_def = PipelineDefinition::Resolved(pipeline.clone());
512        let pipeline_param = GreptimePipelineParams::default();
513
514        (pipeline, schema_info, pipeline_def, pipeline_param)
515    }};
516}
517
518#[cfg(test)]
519mod tests {
520    use std::collections::BTreeMap;
521    use std::sync::Arc;
522
523    use api::v1::Rows;
524    use datatypes::schema::{FulltextOptions, SkippingIndexOptions};
525    use greptime_proto::v1::value::ValueData;
526    use greptime_proto::v1::{self, ColumnDataType, SemanticType};
527    use vrl::prelude::Bytes;
528    use vrl::value::KeyString;
529
530    use super::*;
531
532    #[test]
533    fn test_pipeline_prepare() {
534        let input_value_str = r#"
535                    {
536                        "my_field": "1,2",
537                        "foo": "bar",
538                        "ts": "1"
539                    }
540                "#;
541        let input_value: serde_json::Value = serde_json::from_str(input_value_str).unwrap();
542
543        let pipeline_yaml = r#"description: 'Pipeline for Apache Tomcat'
544processors:
545    - csv:
546        field: my_field
547        target_fields: field1, field2
548    - epoch:
549        field: ts
550        resolution: ns
551transform:
552    - field: field1
553      type: uint32
554    - field: field2
555      type: uint32
556    - field: ts
557      type: timestamp, ns
558      index: time
559    "#;
560
561        let pipeline: Pipeline = parse(&Content::Yaml(pipeline_yaml)).unwrap();
562        let (pipeline, mut schema_info, pipeline_def, pipeline_param) = setup_pipeline!(pipeline);
563        let pipeline_ctx = PipelineContext::new(
564            &pipeline_def,
565            &pipeline_param,
566            session::context::Channel::Unknown,
567        );
568
569        let payload = input_value.into();
570        let mut result = pipeline
571            .exec_mut(payload, &pipeline_ctx, &mut schema_info)
572            .unwrap()
573            .into_transformed()
574            .unwrap();
575
576        let (row, _table_suffix) = result.swap_remove(0);
577        assert_eq!(row.values[0].value_data, Some(ValueData::U32Value(1)));
578        assert_eq!(row.values[1].value_data, Some(ValueData::U32Value(2)));
579        match &row.values[2].value_data {
580            Some(ValueData::TimestampNanosecondValue(v)) => {
581                assert_ne!(v, &0);
582            }
583            _ => panic!("expect null value"),
584        }
585    }
586
587    #[test]
588    fn test_dissect_pipeline() {
589        let message = r#"129.37.245.88 - meln1ks [01/Aug/2024:14:22:47 +0800] "PATCH /observability/metrics/production HTTP/1.0" 501 33085"#.to_string();
590        let pipeline_str = r#"processors:
591    - dissect:
592        fields:
593          - message
594        patterns:
595          - "%{ip} %{?ignored} %{username} [%{ts}] \"%{method} %{path} %{proto}\" %{status} %{bytes}"
596    - date:
597        fields:
598          - ts
599        formats:
600          - "%d/%b/%Y:%H:%M:%S %z"
601
602transform:
603    - fields:
604        - ip
605        - username
606        - method
607        - path
608        - proto
609      type: string
610    - fields:
611        - status
612      type: uint16
613    - fields:
614        - bytes
615      type: uint32
616    - field: ts
617      type: timestamp, ns
618      index: time"#;
619        let pipeline: Pipeline = parse(&Content::Yaml(pipeline_str)).unwrap();
620        let pipeline = Arc::new(pipeline);
621        let schema = pipeline.schemas().unwrap();
622        let mut schema_info = SchemaInfo::from_schema_list(schema.clone());
623
624        let pipeline_def = crate::PipelineDefinition::Resolved(pipeline.clone());
625        let pipeline_param = crate::GreptimePipelineParams::default();
626        let pipeline_ctx = PipelineContext::new(
627            &pipeline_def,
628            &pipeline_param,
629            session::context::Channel::Unknown,
630        );
631        let payload = VrlValue::Object(BTreeMap::from([(
632            KeyString::from("message"),
633            VrlValue::Bytes(Bytes::from(message)),
634        )]));
635
636        let result = pipeline
637            .exec_mut(payload, &pipeline_ctx, &mut schema_info)
638            .unwrap()
639            .into_transformed()
640            .unwrap();
641
642        assert_eq!(schema_info.schema.len(), result[0].0.values.len());
643        let test = [
644            (
645                ColumnDataType::String as i32,
646                Some(ValueData::StringValue("129.37.245.88".into())),
647            ),
648            (
649                ColumnDataType::String as i32,
650                Some(ValueData::StringValue("meln1ks".into())),
651            ),
652            (
653                ColumnDataType::String as i32,
654                Some(ValueData::StringValue("PATCH".into())),
655            ),
656            (
657                ColumnDataType::String as i32,
658                Some(ValueData::StringValue(
659                    "/observability/metrics/production".into(),
660                )),
661            ),
662            (
663                ColumnDataType::String as i32,
664                Some(ValueData::StringValue("HTTP/1.0".into())),
665            ),
666            (
667                ColumnDataType::Uint16 as i32,
668                Some(ValueData::U16Value(501)),
669            ),
670            (
671                ColumnDataType::Uint32 as i32,
672                Some(ValueData::U32Value(33085)),
673            ),
674            (
675                ColumnDataType::TimestampNanosecond as i32,
676                Some(ValueData::TimestampNanosecondValue(1722493367000000000)),
677            ),
678        ];
679        // manually set schema
680        let schema = pipeline.schemas().unwrap();
681        for i in 0..schema.len() {
682            let schema = &schema[i];
683            let value = &result[0].0.values[i];
684            assert_eq!(schema.datatype, test[i].0);
685            assert_eq!(value.value_data, test[i].1);
686        }
687    }
688
689    #[test]
690    fn test_csv_pipeline() {
691        let input_value_str = r#"
692                    {
693                        "my_field": "1,2",
694                        "foo": "bar",
695                        "ts": "1"
696                    }
697                "#;
698        let input_value: serde_json::Value = serde_json::from_str(input_value_str).unwrap();
699
700        let pipeline_yaml = r#"
701    description: Pipeline for Apache Tomcat
702    processors:
703      - csv:
704          field: my_field
705          target_fields: field1, field2
706      - epoch:
707          field: ts
708          resolution: ns
709    transform:
710      - field: field1
711        type: uint32
712      - field: field2
713        type: uint32
714      - field: ts
715        type: timestamp, ns
716        index: time
717    "#;
718
719        let pipeline: Pipeline = parse(&Content::Yaml(pipeline_yaml)).unwrap();
720        let (pipeline, mut schema_info, pipeline_def, pipeline_param) = setup_pipeline!(pipeline);
721        let pipeline_ctx = PipelineContext::new(
722            &pipeline_def,
723            &pipeline_param,
724            session::context::Channel::Unknown,
725        );
726
727        let payload = input_value.into();
728        let result = pipeline
729            .exec_mut(payload, &pipeline_ctx, &mut schema_info)
730            .unwrap()
731            .into_transformed()
732            .unwrap();
733        assert_eq!(
734            result[0].0.values[0].value_data,
735            Some(ValueData::U32Value(1))
736        );
737        assert_eq!(
738            result[0].0.values[1].value_data,
739            Some(ValueData::U32Value(2))
740        );
741        match &result[0].0.values[2].value_data {
742            Some(ValueData::TimestampNanosecondValue(v)) => {
743                assert_ne!(v, &0);
744            }
745            _ => panic!("expect null value"),
746        }
747    }
748
749    #[test]
750    fn test_date_pipeline() {
751        let input_value_str = r#"
752                {
753                    "my_field": "1,2",
754                    "foo": "bar",
755                    "test_time": "2014-5-17T04:34:56+00:00"
756                }
757            "#;
758        let input_value: serde_json::Value = serde_json::from_str(input_value_str).unwrap();
759
760        let pipeline_yaml = r#"---
761description: Pipeline for Apache Tomcat
762
763processors:
764    - date:
765        field: test_time
766
767transform:
768    - field: test_time
769      type: timestamp, ns
770      index: time
771    "#;
772
773        let pipeline: Pipeline = parse(&Content::Yaml(pipeline_yaml)).unwrap();
774        let pipeline = Arc::new(pipeline);
775        let schema = pipeline.schemas().unwrap();
776        let mut schema_info = SchemaInfo::from_schema_list(schema.clone());
777
778        let pipeline_def = crate::PipelineDefinition::Resolved(pipeline.clone());
779        let pipeline_param = crate::GreptimePipelineParams::default();
780        let pipeline_ctx = PipelineContext::new(
781            &pipeline_def,
782            &pipeline_param,
783            session::context::Channel::Unknown,
784        );
785        let schema = pipeline.schemas().unwrap().clone();
786        let result = input_value.into();
787
788        let rows_with_suffix = pipeline
789            .exec_mut(result, &pipeline_ctx, &mut schema_info)
790            .unwrap()
791            .into_transformed()
792            .unwrap();
793        let output = Rows {
794            schema,
795            rows: rows_with_suffix.into_iter().map(|(r, _)| r).collect(),
796        };
797        let schemas = output.schema;
798
799        assert_eq!(schemas.len(), 1);
800        let schema = schemas[0].clone();
801        assert_eq!("test_time", schema.column_name);
802        assert_eq!(ColumnDataType::TimestampNanosecond as i32, schema.datatype);
803        assert_eq!(SemanticType::Timestamp as i32, schema.semantic_type);
804
805        let row = output.rows[0].clone();
806        assert_eq!(1, row.values.len());
807        let value_data = row.values[0].clone().value_data;
808        assert_eq!(
809            Some(v1::value::ValueData::TimestampNanosecondValue(
810                1400301296000000000
811            )),
812            value_data
813        );
814    }
815
816    #[test]
817    fn test_dispatcher() {
818        let pipeline_yaml = r#"
819---
820description: Pipeline for Apache Tomcat
821
822processors:
823  - epoch:
824      field: ts
825      resolution: ns
826
827dispatcher:
828  field: typename
829  rules:
830    - value: http
831      table_suffix: http_events
832    - value: database
833      table_suffix: db_events
834      pipeline: database_pipeline
835
836transform:
837  - field: typename
838    type: string
839  - field: ts
840    type: timestamp, ns
841    index: time
842"#;
843        let pipeline: Pipeline = parse(&Content::Yaml(pipeline_yaml)).unwrap();
844        let dispatcher = pipeline.dispatcher.expect("expect dispatcher");
845        assert_eq!(dispatcher.field, "typename");
846
847        assert_eq!(dispatcher.rules.len(), 2);
848
849        assert_eq!(
850            dispatcher.rules[0],
851            crate::dispatcher::Rule {
852                value: VrlValue::Bytes(Bytes::from("http")),
853                table_suffix: "http_events".to_string(),
854                pipeline: None
855            }
856        );
857
858        assert_eq!(
859            dispatcher.rules[1],
860            crate::dispatcher::Rule {
861                value: VrlValue::Bytes(Bytes::from("database")),
862                table_suffix: "db_events".to_string(),
863                pipeline: Some("database_pipeline".to_string()),
864            }
865        );
866
867        let bad_yaml1 = r#"
868---
869description: Pipeline for Apache Tomcat
870
871processors:
872  - epoch:
873      field: ts
874      resolution: ns
875
876dispatcher:
877  _field: typename
878  rules:
879    - value: http
880      table_suffix: http_events
881    - value: database
882      table_suffix: db_events
883      pipeline: database_pipeline
884
885transform:
886  - field: typename
887    type: string
888  - field: ts
889    type: timestamp, ns
890    index: time
891"#;
892        let bad_yaml2 = r#"
893---
894description: Pipeline for Apache Tomcat
895
896processors:
897  - epoch:
898      field: ts
899      resolution: ns
900dispatcher:
901  field: typename
902  rules:
903    - value: http
904      _table_suffix: http_events
905    - value: database
906      _table_suffix: db_events
907      pipeline: database_pipeline
908
909transform:
910  - field: typename
911    type: string
912  - field: ts
913    type: timestamp, ns
914    index: time
915"#;
916        let bad_yaml3 = r#"
917---
918description: Pipeline for Apache Tomcat
919
920processors:
921  - epoch:
922      field: ts
923      resolution: ns
924dispatcher:
925  field: typename
926  rules:
927    - _value: http
928      table_suffix: http_events
929    - _value: database
930      table_suffix: db_events
931      pipeline: database_pipeline
932
933transform:
934  - field: typename
935    type: string
936  - field: ts
937    type: timestamp, ns
938    index: time
939"#;
940
941        let r: Result<Pipeline> = parse(&Content::Yaml(bad_yaml1));
942        assert!(r.is_err());
943        let r: Result<Pipeline> = parse(&Content::Yaml(bad_yaml2));
944        assert!(r.is_err());
945        let r: Result<Pipeline> = parse(&Content::Yaml(bad_yaml3));
946        assert!(r.is_err());
947    }
948
949    /// Test one-to-many VRL pipeline expansion.
950    /// A VRL processor can return an array, which results in multiple output rows.
951    #[test]
952    fn test_one_to_many_vrl_expansion() {
953        let pipeline_yaml = r#"
954processors:
955  - epoch:
956      field: timestamp
957      resolution: ms
958  - vrl:
959      source: |
960        events = del(.events)
961        base_host = del(.host)
962        base_ts = del(.timestamp)
963        map_values(array!(events)) -> |event| {
964            {
965                "host": base_host,
966                "event_type": event.type,
967                "event_value": event.value,
968                "timestamp": base_ts
969            }
970        }
971
972transform:
973  - field: host
974    type: string
975  - field: event_type
976    type: string
977  - field: event_value
978    type: int32
979  - field: timestamp
980    type: timestamp, ms
981    index: time
982"#;
983
984        let pipeline: Pipeline = parse(&Content::Yaml(pipeline_yaml)).unwrap();
985        let (pipeline, mut schema_info, pipeline_def, pipeline_param) = setup_pipeline!(pipeline);
986        let pipeline_ctx = PipelineContext::new(
987            &pipeline_def,
988            &pipeline_param,
989            session::context::Channel::Unknown,
990        );
991
992        // Input with 3 events
993        let input_value: serde_json::Value = serde_json::from_str(
994            r#"{
995                "host": "server1",
996                "timestamp": 1716668197217,
997                "events": [
998                    {"type": "cpu", "value": 80},
999                    {"type": "memory", "value": 60},
1000                    {"type": "disk", "value": 45}
1001                ]
1002            }"#,
1003        )
1004        .unwrap();
1005
1006        let payload = input_value.into();
1007        let result = pipeline
1008            .exec_mut(payload, &pipeline_ctx, &mut schema_info)
1009            .unwrap()
1010            .into_transformed()
1011            .unwrap();
1012
1013        // Should produce 3 rows from 1 input
1014        assert_eq!(result.len(), 3);
1015
1016        // Verify each row has correct structure
1017        for (row, _table_suffix) in &result {
1018            assert_eq!(row.values.len(), 4); // host, event_type, event_value, timestamp
1019            // First value should be "server1"
1020            assert_eq!(
1021                row.values[0].value_data,
1022                Some(ValueData::StringValue("server1".to_string()))
1023            );
1024            // Last value should be the timestamp
1025            assert_eq!(
1026                row.values[3].value_data,
1027                Some(ValueData::TimestampMillisecondValue(1716668197217))
1028            );
1029        }
1030
1031        // Verify event types
1032        let event_types: Vec<_> = result
1033            .iter()
1034            .map(|(r, _)| match &r.values[1].value_data {
1035                Some(ValueData::StringValue(s)) => s.clone(),
1036                _ => panic!("expected string"),
1037            })
1038            .collect();
1039        assert!(event_types.contains(&"cpu".to_string()));
1040        assert!(event_types.contains(&"memory".to_string()));
1041        assert!(event_types.contains(&"disk".to_string()));
1042    }
1043
1044    /// Test that single object output still works (backward compatibility)
1045    #[test]
1046    fn test_single_object_output_unchanged() {
1047        let pipeline_yaml = r#"
1048processors:
1049  - epoch:
1050      field: ts
1051      resolution: ms
1052  - vrl:
1053      source: |
1054        .processed = true
1055        .
1056
1057transform:
1058  - field: name
1059    type: string
1060  - field: processed
1061    type: boolean
1062  - field: ts
1063    type: timestamp, ms
1064    index: time
1065"#;
1066
1067        let pipeline: Pipeline = parse(&Content::Yaml(pipeline_yaml)).unwrap();
1068        let (pipeline, mut schema_info, pipeline_def, pipeline_param) = setup_pipeline!(pipeline);
1069        let pipeline_ctx = PipelineContext::new(
1070            &pipeline_def,
1071            &pipeline_param,
1072            session::context::Channel::Unknown,
1073        );
1074
1075        let input_value: serde_json::Value = serde_json::from_str(
1076            r#"{
1077                "name": "test",
1078                "ts": 1716668197217
1079            }"#,
1080        )
1081        .unwrap();
1082
1083        let payload = input_value.into();
1084        let result = pipeline
1085            .exec_mut(payload, &pipeline_ctx, &mut schema_info)
1086            .unwrap()
1087            .into_transformed()
1088            .unwrap();
1089
1090        // Should produce exactly 1 row
1091        assert_eq!(result.len(), 1);
1092        assert_eq!(
1093            result[0].0.values[0].value_data,
1094            Some(ValueData::StringValue("test".to_string()))
1095        );
1096        assert_eq!(
1097            result[0].0.values[1].value_data,
1098            Some(ValueData::BoolValue(true))
1099        );
1100    }
1101
1102    /// Test that empty array produces zero rows
1103    #[test]
1104    fn test_empty_array_produces_zero_rows() {
1105        let pipeline_yaml = r#"
1106processors:
1107  - vrl:
1108      source: |
1109        .events
1110
1111transform:
1112  - field: value
1113    type: int32
1114  - field: greptime_timestamp
1115    type: timestamp, ns
1116    index: time
1117"#;
1118
1119        let pipeline: Pipeline = parse(&Content::Yaml(pipeline_yaml)).unwrap();
1120        let (pipeline, mut schema_info, pipeline_def, pipeline_param) = setup_pipeline!(pipeline);
1121        let pipeline_ctx = PipelineContext::new(
1122            &pipeline_def,
1123            &pipeline_param,
1124            session::context::Channel::Unknown,
1125        );
1126
1127        let input_value: serde_json::Value = serde_json::from_str(r#"{"events": []}"#).unwrap();
1128
1129        let payload = input_value.into();
1130        let result = pipeline
1131            .exec_mut(payload, &pipeline_ctx, &mut schema_info)
1132            .unwrap()
1133            .into_transformed()
1134            .unwrap();
1135
1136        // Empty array should produce zero rows
1137        assert_eq!(result.len(), 0);
1138    }
1139
1140    /// Test that array elements must be objects
1141    #[test]
1142    fn test_array_element_must_be_object() {
1143        let pipeline_yaml = r#"
1144processors:
1145  - vrl:
1146      source: |
1147        .items
1148
1149transform:
1150  - field: value
1151    type: int32
1152  - field: greptime_timestamp
1153    type: timestamp, ns
1154    index: time
1155"#;
1156
1157        let pipeline: Pipeline = parse(&Content::Yaml(pipeline_yaml)).unwrap();
1158        let (pipeline, mut schema_info, pipeline_def, pipeline_param) = setup_pipeline!(pipeline);
1159        let pipeline_ctx = PipelineContext::new(
1160            &pipeline_def,
1161            &pipeline_param,
1162            session::context::Channel::Unknown,
1163        );
1164
1165        // Array with non-object elements should fail
1166        let input_value: serde_json::Value =
1167            serde_json::from_str(r#"{"items": [1, 2, 3]}"#).unwrap();
1168
1169        let payload = input_value.into();
1170        let result = pipeline.exec_mut(payload, &pipeline_ctx, &mut schema_info);
1171
1172        assert!(result.is_err());
1173        let err_msg = result.unwrap_err().to_string();
1174        assert!(
1175            err_msg.contains("must be an object"),
1176            "Expected error about non-object element, got: {}",
1177            err_msg
1178        );
1179    }
1180
1181    /// Test one-to-many with table suffix from VRL hint
1182    #[test]
1183    fn test_one_to_many_with_table_suffix_hint() {
1184        let pipeline_yaml = r#"
1185processors:
1186  - epoch:
1187      field: ts
1188      resolution: ms
1189  - vrl:
1190      source: |
1191        .greptime_table_suffix = "_" + string!(.category)
1192        .
1193
1194transform:
1195  - field: name
1196    type: string
1197  - field: category
1198    type: string
1199  - field: ts
1200    type: timestamp, ms
1201    index: time
1202"#;
1203
1204        let pipeline: Pipeline = parse(&Content::Yaml(pipeline_yaml)).unwrap();
1205        let (pipeline, mut schema_info, pipeline_def, pipeline_param) = setup_pipeline!(pipeline);
1206        let pipeline_ctx = PipelineContext::new(
1207            &pipeline_def,
1208            &pipeline_param,
1209            session::context::Channel::Unknown,
1210        );
1211
1212        let input_value: serde_json::Value = serde_json::from_str(
1213            r#"{
1214                "name": "test",
1215                "category": "metrics",
1216                "ts": 1716668197217
1217            }"#,
1218        )
1219        .unwrap();
1220
1221        let payload = input_value.into();
1222        let result = pipeline
1223            .exec_mut(payload, &pipeline_ctx, &mut schema_info)
1224            .unwrap()
1225            .into_transformed()
1226            .unwrap();
1227
1228        // Should have table suffix extracted per row
1229        assert_eq!(result.len(), 1);
1230        assert_eq!(result[0].1, Some("_metrics".to_string()));
1231    }
1232
1233    /// Test one-to-many with per-row table suffix
1234    #[test]
1235    fn test_one_to_many_per_row_table_suffix() {
1236        let pipeline_yaml = r#"
1237processors:
1238  - epoch:
1239      field: timestamp
1240      resolution: ms
1241  - vrl:
1242      source: |
1243        events = del(.events)
1244        base_ts = del(.timestamp)
1245
1246        map_values(array!(events)) -> |event| {
1247            suffix = "_" + string!(event.category)
1248            {
1249                "name": event.name,
1250                "value": event.value,
1251                "timestamp": base_ts,
1252                "greptime_table_suffix": suffix
1253            }
1254        }
1255
1256transform:
1257  - field: name
1258    type: string
1259  - field: value
1260    type: int32
1261  - field: timestamp
1262    type: timestamp, ms
1263    index: time
1264"#;
1265
1266        let pipeline: Pipeline = parse(&Content::Yaml(pipeline_yaml)).unwrap();
1267        let (pipeline, mut schema_info, pipeline_def, pipeline_param) = setup_pipeline!(pipeline);
1268        let pipeline_ctx = PipelineContext::new(
1269            &pipeline_def,
1270            &pipeline_param,
1271            session::context::Channel::Unknown,
1272        );
1273
1274        // Input with events that should go to different tables
1275        let input_value: serde_json::Value = serde_json::from_str(
1276            r#"{
1277                "timestamp": 1716668197217,
1278                "events": [
1279                    {"name": "cpu_usage", "value": 80, "category": "cpu"},
1280                    {"name": "mem_usage", "value": 60, "category": "memory"},
1281                    {"name": "cpu_temp", "value": 45, "category": "cpu"}
1282                ]
1283            }"#,
1284        )
1285        .unwrap();
1286
1287        let payload = input_value.into();
1288        let result = pipeline
1289            .exec_mut(payload, &pipeline_ctx, &mut schema_info)
1290            .unwrap()
1291            .into_transformed()
1292            .unwrap();
1293
1294        // Should produce 3 rows
1295        assert_eq!(result.len(), 3);
1296
1297        // Collect table suffixes
1298        let table_suffixes: Vec<_> = result.iter().map(|(_, suffix)| suffix.clone()).collect();
1299
1300        // Should have different table suffixes per row
1301        assert!(table_suffixes.contains(&Some("_cpu".to_string())));
1302        assert!(table_suffixes.contains(&Some("_memory".to_string())));
1303
1304        // Count rows per table suffix
1305        let cpu_count = table_suffixes
1306            .iter()
1307            .filter(|s| *s == &Some("_cpu".to_string()))
1308            .count();
1309        let memory_count = table_suffixes
1310            .iter()
1311            .filter(|s| *s == &Some("_memory".to_string()))
1312            .count();
1313        assert_eq!(cpu_count, 2);
1314        assert_eq!(memory_count, 1);
1315    }
1316
1317    /// Test that one-to-many mapping preserves per-row ContextOpt in HashMap
1318    #[test]
1319    fn test_one_to_many_hashmap_contextopt_preservation() {
1320        let pipeline_yaml = r#"
1321processors:
1322  - epoch:
1323      field: timestamp
1324      resolution: ms
1325  - vrl:
1326      source: |
1327        events = del(.events)
1328        base_ts = del(.timestamp)
1329
1330        map_values(array!(events)) -> |event| {
1331            # Set different TTL values per event type
1332            ttl = if event.type == "critical" {
1333                "1h"
1334            } else if event.type == "warning" {
1335                "24h"
1336            } else {
1337                "7d"
1338            }
1339
1340            {
1341                "host": del(.host),
1342                "event_type": event.type,
1343                "event_value": event.value,
1344                "timestamp": base_ts,
1345                "greptime_ttl": ttl
1346            }
1347        }
1348
1349transform:
1350  - field: host
1351    type: string
1352  - field: event_type
1353    type: string
1354  - field: event_value
1355    type: int32
1356  - field: timestamp
1357    type: timestamp, ms
1358    index: time
1359"#;
1360
1361        let pipeline: Pipeline = parse(&Content::Yaml(pipeline_yaml)).unwrap();
1362        let (pipeline, mut schema_info, pipeline_def, pipeline_param) = setup_pipeline!(pipeline);
1363        let pipeline_ctx = PipelineContext::new(
1364            &pipeline_def,
1365            &pipeline_param,
1366            session::context::Channel::Unknown,
1367        );
1368
1369        // Input with events that should have different ContextOpt values
1370        let input_value: serde_json::Value = serde_json::from_str(
1371            r#"{
1372                "host": "server1",
1373                "timestamp": 1716668197217,
1374                "events": [
1375                    {"type": "critical", "value": 100},
1376                    {"type": "warning", "value": 50},
1377                    {"type": "info", "value": 25}
1378                ]
1379            }"#,
1380        )
1381        .unwrap();
1382
1383        let payload = input_value.into();
1384        let result = pipeline
1385            .exec_mut(payload, &pipeline_ctx, &mut schema_info)
1386            .unwrap();
1387
1388        // Extract the HashMap structure
1389        let rows_by_context = result.into_transformed_hashmap().unwrap();
1390
1391        // Should have 3 different ContextOpt groups due to different TTL values
1392        assert_eq!(rows_by_context.len(), 3);
1393
1394        // Verify each ContextOpt group has exactly 1 row and different configurations
1395        let mut context_opts = Vec::new();
1396        for (opt, rows) in &rows_by_context {
1397            assert_eq!(rows.len(), 1); // Each group should have exactly 1 row
1398            context_opts.push(opt.clone());
1399        }
1400
1401        // ContextOpts should be different due to different TTL values
1402        assert_ne!(context_opts[0], context_opts[1]);
1403        assert_ne!(context_opts[1], context_opts[2]);
1404        assert_ne!(context_opts[0], context_opts[2]);
1405
1406        // Verify the rows are correctly structured
1407        for rows in rows_by_context.values() {
1408            for (row, _table_suffix) in rows {
1409                assert_eq!(row.values.len(), 4); // host, event_type, event_value, timestamp
1410            }
1411        }
1412    }
1413
1414    /// Test that single object input still works with HashMap structure
1415    #[test]
1416    fn test_single_object_hashmap_compatibility() {
1417        let pipeline_yaml = r#"
1418processors:
1419  - epoch:
1420      field: ts
1421      resolution: ms
1422  - vrl:
1423      source: |
1424        .processed = true
1425        .
1426
1427transform:
1428  - field: name
1429    type: string
1430  - field: processed
1431    type: boolean
1432  - field: ts
1433    type: timestamp, ms
1434    index: time
1435"#;
1436
1437        let pipeline: Pipeline = parse(&Content::Yaml(pipeline_yaml)).unwrap();
1438        let (pipeline, mut schema_info, pipeline_def, pipeline_param) = setup_pipeline!(pipeline);
1439        let pipeline_ctx = PipelineContext::new(
1440            &pipeline_def,
1441            &pipeline_param,
1442            session::context::Channel::Unknown,
1443        );
1444
1445        let input_value: serde_json::Value = serde_json::from_str(
1446            r#"{
1447                "name": "test",
1448                "ts": 1716668197217
1449            }"#,
1450        )
1451        .unwrap();
1452
1453        let payload = input_value.into();
1454        let result = pipeline
1455            .exec_mut(payload, &pipeline_ctx, &mut schema_info)
1456            .unwrap();
1457
1458        // Extract the HashMap structure
1459        let rows_by_context = result.into_transformed_hashmap().unwrap();
1460
1461        // Single object should produce exactly 1 ContextOpt group
1462        assert_eq!(rows_by_context.len(), 1);
1463
1464        let (_opt, rows) = rows_by_context.into_iter().next().unwrap();
1465        assert_eq!(rows.len(), 1);
1466
1467        // Verify the row structure
1468        let (row, _table_suffix) = &rows[0];
1469        assert_eq!(row.values.len(), 3); // name, processed, timestamp
1470    }
1471
1472    /// Test that empty arrays work correctly with HashMap structure
1473    #[test]
1474    fn test_empty_array_hashmap() {
1475        let pipeline_yaml = r#"
1476processors:
1477  - vrl:
1478      source: |
1479        .events
1480
1481transform:
1482  - field: value
1483    type: int32
1484  - field: greptime_timestamp
1485    type: timestamp, ns
1486    index: time
1487"#;
1488
1489        let pipeline: Pipeline = parse(&Content::Yaml(pipeline_yaml)).unwrap();
1490        let (pipeline, mut schema_info, pipeline_def, pipeline_param) = setup_pipeline!(pipeline);
1491        let pipeline_ctx = PipelineContext::new(
1492            &pipeline_def,
1493            &pipeline_param,
1494            session::context::Channel::Unknown,
1495        );
1496
1497        let input_value: serde_json::Value = serde_json::from_str(r#"{"events": []}"#).unwrap();
1498
1499        let payload = input_value.into();
1500        let result = pipeline
1501            .exec_mut(payload, &pipeline_ctx, &mut schema_info)
1502            .unwrap();
1503
1504        // Extract the HashMap structure
1505        let rows_by_context = result.into_transformed_hashmap().unwrap();
1506
1507        // Empty array should produce empty HashMap
1508        assert_eq!(rows_by_context.len(), 0);
1509    }
1510
1511    #[test]
1512    fn test_pipeline_detailed_index_options_roundtrip() {
1513        let pipeline_yaml = r#"
1514transform:
1515  - field: message
1516    type: string
1517    index:
1518      type: fulltext
1519      options:
1520        analyzer: Chinese
1521        case_sensitive: true
1522        backend: tantivy
1523  - field: trace_id
1524    type: int64
1525    index:
1526      type: skipping
1527      options:
1528        granularity: 2048
1529        false_positive_rate: 0.02
1530        type: BLOOM
1531  - field: ts
1532    type: timestamp, ns
1533    index: time
1534"#;
1535
1536        let pipeline: Pipeline = parse(&Content::Yaml(pipeline_yaml)).unwrap();
1537        let schema = pipeline.schemas().unwrap().clone();
1538
1539        let message = schema
1540            .iter()
1541            .find(|column| column.column_name == "message")
1542            .unwrap();
1543        let trace_id = schema
1544            .iter()
1545            .find(|column| column.column_name == "trace_id")
1546            .unwrap();
1547        let message_options = message.options.clone();
1548        let trace_id_options = trace_id.options.clone();
1549
1550        let fulltext: FulltextOptions = serde_json::from_str(
1551            message
1552                .options
1553                .as_ref()
1554                .unwrap()
1555                .options
1556                .get("fulltext")
1557                .unwrap(),
1558        )
1559        .unwrap();
1560        assert!(fulltext.enable);
1561        assert_eq!(fulltext.analyzer.to_string(), "Chinese");
1562        assert!(fulltext.case_sensitive);
1563        assert_eq!(fulltext.backend.to_string(), "tantivy");
1564
1565        let skipping: SkippingIndexOptions = serde_json::from_str(
1566            trace_id
1567                .options
1568                .as_ref()
1569                .unwrap()
1570                .options
1571                .get("skipping_index")
1572                .unwrap(),
1573        )
1574        .unwrap();
1575        assert_eq!(skipping.granularity, 2048);
1576        assert_eq!(skipping.false_positive_rate(), 0.02);
1577        assert_eq!(skipping.index_type.to_string(), "BLOOM");
1578
1579        let roundtrip_schema = SchemaInfo::from_schema_list(schema)
1580            .column_schemas()
1581            .unwrap();
1582        let roundtrip_message = roundtrip_schema
1583            .iter()
1584            .find(|column| column.column_name == "message")
1585            .unwrap();
1586        let roundtrip_trace_id = roundtrip_schema
1587            .iter()
1588            .find(|column| column.column_name == "trace_id")
1589            .unwrap();
1590
1591        assert_eq!(message_options, roundtrip_message.options);
1592        assert_eq!(trace_id_options, roundtrip_trace_id.options);
1593    }
1594}
pipeline/etl.rs

pipeline/
etl.rs
