operator/insert/
otlp.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//! Side-effect-free destination validation for prepared OTLP batches. Catalog
16//! failures propagate; row errors let the converter reject complete source-point fan-outs.
17
18use std::collections::HashSet;
19
20use ahash::{HashMap as AHashMap, HashMapExt};
21use api::helper::{ColumnDataTypeWrapper, to_grpc_value};
22use api::v1::alter_table_expr::Kind;
23use api::v1::column_def::try_as_column_schema;
24use api::v1::{
25    ColumnDataType, ColumnDef, ColumnSchema, Row, RowInsertRequest, Rows, SemanticType, Value,
26};
27use common_query::native_histogram::native_histogram_value_type;
28use common_query::prelude::{greptime_timestamp, greptime_value};
29use common_time::timestamp::TimeUnit;
30use datatypes::schema::ColumnSchema as DataColumn;
31use session::context::QueryContextRef;
32use snafu::prelude::*;
33use store_api::metric_engine_consts::{
34    LOGICAL_TABLE_METADATA_KEY, METRIC_ENGINE_NAME, PHYSICAL_TABLE_METADATA_KEY,
35};
36use table::TableRef;
37
38use crate::error::{self, InvalidInsertRequestSnafu, Result};
39use crate::insert::{
40    AutoCreateTableType, Inserter, convert_rows_time_unit, convert_timestamp_value_data,
41    request_is_native_histogram, table_is_native_histogram, table_time_index_unit,
42};
43use crate::req_convert::insert::{ImpureDefaultFiller, value_matches_type};
44
45pub struct MetricInsertPreflight {
46    /// Rejections in batch order, then row order.
47    pub rejections: Vec<Vec<Option<String>>>,
48    /// Backing tables for existing or newly created logical metric-engine destinations.
49    pub physical_tables: std::collections::HashMap<String, String>,
50}
51
52struct PhysicalColumns {
53    by_name: AHashMap<String, (DataColumn, SemanticType)>,
54    fields: Vec<String>,
55}
56
57impl PhysicalColumns {
58    fn new(columns: Vec<(DataColumn, SemanticType)>) -> Self {
59        let fields = columns
60            .iter()
61            .filter(|(_, semantic)| *semantic == SemanticType::Field)
62            .map(|(column, _)| column.name.clone())
63            .collect();
64        let by_name = columns
65            .into_iter()
66            .map(|(column, semantic)| (column.name.clone(), (column, semantic)))
67            .collect();
68        Self { by_name, fields }
69    }
70}
71
72impl Inserter {
73    /// Validate each row independently, sharing schema planning within each batch.
74    /// Fresh cross-point conflicts are resolved by the converter after this pass.
75    pub async fn validate_and_normalize_otlp_metrics(
76        &self,
77        batches: &mut [RowInsertRequest],
78        ctx: &QueryContextRef,
79        metric_engine: bool,
80        physical_table: &str,
81    ) -> Result<MetricInsertPreflight> {
82        let disabled = self.auto_create_disabled_reason(ctx)?;
83        let mut tables = AHashMap::new();
84        let mut physical_tables = std::collections::HashMap::new();
85        for request in batches.iter() {
86            if !tables.contains_key(&request.table_name) {
87                let table = self
88                    .get_table(
89                        ctx.current_catalog(),
90                        &ctx.current_schema(),
91                        &request.table_name,
92                    )
93                    .await?;
94                tables.insert(request.table_name.clone(), table);
95            }
96            let backing = match tables[&request.table_name].as_ref() {
97                Some(table) if table.table_info().meta.engine == METRIC_ENGINE_NAME => table
98                    .table_info()
99                    .meta
100                    .options
101                    .extra_options
102                    .get(LOGICAL_TABLE_METADATA_KEY)
103                    .cloned(),
104                None if metric_engine => Some(physical_table.to_string()),
105                _ => None,
106            };
107            if let Some(backing) = backing {
108                physical_tables.insert(request.table_name.clone(), backing);
109            }
110        }
111        // Shared schemas accumulate tags across logical tables; index them once per request.
112        let mut physical_columns = AHashMap::new();
113        for backing in physical_tables.values() {
114            if !tables.contains_key(backing) {
115                let table = self
116                    .get_table(ctx.current_catalog(), &ctx.current_schema(), backing)
117                    .await?;
118                tables.insert(backing.clone(), table);
119            }
120            physical_columns.entry(backing.as_str()).or_insert_with(|| {
121                PhysicalColumns::new(
122                    tables[backing]
123                        .as_ref()
124                        .map(table_columns)
125                        .unwrap_or_else(default_physical_columns),
126                )
127            });
128        }
129        // Match ordinary insertion: impure defaults are evaluated once per table
130        // and request, while supplied nulls remain supplied nulls.
131        let mut fillers = AHashMap::new();
132        let mut build_plan = |source: &mut RowInsertRequest| {
133            let rows = source.rows.as_mut().context(InvalidInsertRequestSnafu {
134                reason: "missing metric rows",
135            })?;
136            let input_columns = rows.schema.len();
137            let source_timestamp = timestamp_column(&rows.schema);
138            let request = &mut RowInsertRequest {
139                table_name: source.table_name.clone(),
140                rows: Some(Rows {
141                    schema: std::mem::take(&mut rows.schema),
142                    rows: Vec::new(),
143                }),
144            };
145            let auto_create = disabled.is_none()
146                || Self::is_auto_create_exempt_private_table(
147                    &ctx.current_schema(),
148                    &request.table_name,
149                );
150            let target = tables[&request.table_name].as_ref();
151            let backing = physical_tables.get(&request.table_name);
152            let physical = backing.and_then(|name| tables[name].as_ref());
153            if let Some(physical) = physical {
154                ensure!(
155                    physical
156                        .table_info()
157                        .meta
158                        .options
159                        .extra_options
160                        .contains_key(PHYSICAL_TABLE_METADATA_KEY),
161                    InvalidInsertRequestSnafu {
162                        reason: "selected backing table is not a physical metric table"
163                    }
164                );
165            }
166            if metric_engine {
167                ensure!(
168                    backing.is_none() || physical.is_some() || auto_create,
169                    InvalidInsertRequestSnafu {
170                        reason: format!(
171                            "missing physical table and {}",
172                            disabled.unwrap_or_default()
173                        ),
174                    }
175                );
176                let unit = target
177                    .and_then(table_time_index_unit)
178                    .or_else(|| physical.and_then(table_time_index_unit))
179                    .unwrap_or(TimeUnit::Millisecond);
180                if let Some(rows) = &mut request.rows {
181                    convert_rows_time_unit(rows, unit)?;
182                }
183            }
184            let mut columns = if let Some(table) = target {
185                ensure!(
186                    !table
187                        .table_info()
188                        .meta
189                        .options
190                        .extra_options
191                        .contains_key(PHYSICAL_TABLE_METADATA_KEY),
192                    InvalidInsertRequestSnafu {
193                        reason: "direct writes to a physical metric table are forbidden",
194                    }
195                );
196                ensure!(
197                    request_is_native_histogram(&request.rows.as_ref().unwrap().schema)
198                        == table_is_native_histogram(table),
199                    InvalidInsertRequestSnafu {
200                        reason: format!(
201                            "Table `{}` cannot mix native histogram and float sample fields",
202                            request.table_name
203                        ),
204                    }
205                );
206                let mut columns = table_columns(table);
207                if auto_create {
208                    // Ordinary OTLP insertion does not rename columns.
209                    if let Some(alter) = self.get_alter_table_expr_on_demand(
210                        request,
211                        table,
212                        ctx,
213                        metric_engine,
214                        true,
215                        true,
216                    )? && let Some(Kind::AddColumns(add)) = alter.kind
217                    {
218                        for column in add.add_columns {
219                            let def =
220                                column
221                                    .column_def
222                                    .with_context(|| InvalidInsertRequestSnafu {
223                                        reason: "missing column definition",
224                                    })?;
225                            columns.push(column_definition(&def)?);
226                        }
227                    }
228                }
229                columns
230            } else {
231                ensure!(
232                    auto_create,
233                    InvalidInsertRequestSnafu {
234                        reason: format!(
235                            "missing table `{}` and {}",
236                            request.table_name,
237                            disabled.unwrap_or_default()
238                        ),
239                    }
240                );
241                let create_type = if metric_engine {
242                    AutoCreateTableType::Logical(physical_table.to_string())
243                } else {
244                    AutoCreateTableType::Physical
245                };
246                let create =
247                    self.get_create_table_expr_on_demand(request, &create_type, ctx, None)?;
248                if metric_engine {
249                    ensure!(
250                        create
251                            .column_defs
252                            .iter()
253                            .filter(|c| c.semantic_type == SemanticType::Field as i32)
254                            .count()
255                            == 1,
256                        InvalidInsertRequestSnafu {
257                            reason: "logical metrics require exactly one sample field"
258                        }
259                    );
260                }
261                create
262                    .column_defs
263                    .iter()
264                    .map(column_definition)
265                    .collect::<Result<Vec<_>>>()?
266            };
267            let rows = request.rows.as_mut().context(InvalidInsertRequestSnafu {
268                reason: "missing metric rows",
269            })?;
270            if let Some(backing) = backing {
271                // Logical tag nullability follows the shared physical schema.
272                for (column, semantic) in &mut columns {
273                    if *semantic == SemanticType::Tag {
274                        column.set_nullable();
275                    }
276                }
277                validate_physical_columns(rows, &physical_columns[backing.as_str()], auto_create)?;
278            }
279            // Only defaults need values in the planning template.
280            rows.rows.push(Row::default());
281            if let Some(table) = target {
282                if !fillers.contains_key(&request.table_name) {
283                    fillers.insert(
284                        request.table_name.clone(),
285                        ImpureDefaultFiller::new(table.table_info())?,
286                    );
287                }
288                fillers[&request.table_name].fill_rows(rows);
289            }
290            RowValidationPlan::new(
291                input_columns,
292                source_timestamp,
293                std::mem::take(rows),
294                &columns,
295            )
296        };
297        let mut rejections = Vec::with_capacity(batches.len());
298        for request in batches {
299            let num_rows = request
300                .rows
301                .as_ref()
302                .context(InvalidInsertRequestSnafu {
303                    reason: "missing metric rows",
304                })?
305                .rows
306                .len();
307            let plan = match build_plan(request) {
308                Ok(plan) => plan,
309                Err(err) if attributable(&err) => {
310                    rejections.push(vec![Some(err.to_string()); num_rows]);
311                    continue;
312                }
313                Err(err) => return Err(err),
314            };
315            let rows = request.rows.as_mut().context(InvalidInsertRequestSnafu {
316                reason: "missing metric rows",
317            })?;
318            let mut errors = Vec::with_capacity(num_rows);
319            for row in &mut rows.rows {
320                match plan.apply_row(row) {
321                    Ok(()) => errors.push(None),
322                    Err(err) if attributable(&err) => errors.push(Some(err.to_string())),
323                    Err(err) => return Err(err),
324                }
325            }
326            rows.schema = plan.schema;
327            rejections.push(errors);
328        }
329        Ok(MetricInsertPreflight {
330            rejections,
331            physical_tables,
332        })
333    }
334}
335
336fn attributable(err: &error::Error) -> bool {
337    matches!(
338        err,
339        error::Error::InvalidInsertRequest { .. }
340            | error::Error::BuildCreateExprOnInsertion { .. }
341            | error::Error::FindNewColumnsOnInsertion { .. }
342            | error::Error::InvalidColumnDef { .. }
343    )
344}
345
346fn invalid(reason: impl Into<String>) -> error::Error {
347    InvalidInsertRequestSnafu {
348        reason: reason.into(),
349    }
350    .build()
351}
352
353fn column_definition(def: &ColumnDef) -> Result<(DataColumn, SemanticType)> {
354    let column = try_as_column_schema(def).map_err(|err| invalid(err.to_string()))?;
355    let semantic =
356        SemanticType::try_from(def.semantic_type).map_err(|err| invalid(err.to_string()))?;
357    Ok((column, semantic))
358}
359
360fn table_columns(table: &TableRef) -> Vec<(DataColumn, SemanticType)> {
361    let info = table.table_info();
362    let primary_keys: HashSet<_> = info.meta.primary_key_indices.iter().copied().collect();
363    info.meta
364        .schema
365        .column_schemas()
366        .iter()
367        .enumerate()
368        .map(|(index, column)| {
369            let semantic = if column.is_time_index() {
370                SemanticType::Timestamp
371            } else if primary_keys.contains(&index) {
372                SemanticType::Tag
373            } else {
374                SemanticType::Field
375            };
376            (column.clone(), semantic)
377        })
378        .collect()
379}
380
381fn default_physical_columns() -> Vec<(DataColumn, SemanticType)> {
382    use datatypes::prelude::ConcreteDataType;
383    vec![
384        (
385            DataColumn::new(
386                greptime_timestamp(),
387                ConcreteDataType::timestamp_millisecond_datatype(),
388                false,
389            )
390            .with_time_index(true),
391            SemanticType::Timestamp,
392        ),
393        (
394            DataColumn::new(greptime_value(), ConcreteDataType::float64_datatype(), true),
395            SemanticType::Field,
396        ),
397    ]
398}
399
400fn check_definition(
401    source: &ColumnSchema,
402    target: &DataColumn,
403    semantic: SemanticType,
404) -> Result<()> {
405    ensure!(
406        api::helper::is_column_type_value_eq(
407            source.datatype,
408            source.datatype_extension.clone(),
409            &target.data_type
410        ) && source.semantic_type == semantic as i32,
411        InvalidInsertRequestSnafu {
412            reason: format!(
413                "incompatible datatype or semantic type for column `{}`",
414                source.column_name
415            ),
416        }
417    );
418    Ok(())
419}
420
421fn validate_physical_columns(
422    rows: &Rows,
423    columns: &PhysicalColumns,
424    auto_create: bool,
425) -> Result<()> {
426    for source in &rows.schema {
427        if let Some((target, semantic)) = columns.by_name.get(&source.column_name) {
428            check_definition(source, target, *semantic)?;
429        } else {
430            ensure!(
431                auto_create,
432                InvalidInsertRequestSnafu {
433                    reason: format!(
434                        "missing physical column `{}` while auto-DDL is disabled",
435                        source.column_name
436                    )
437                }
438            );
439            ensure!(
440                source.semantic_type == SemanticType::Tag as i32
441                    || (source.semantic_type == SemanticType::Field as i32
442                        && api::helper::is_column_type_value_eq(
443                            source.datatype,
444                            source.datatype_extension.clone(),
445                            native_histogram_value_type()
446                        )),
447                InvalidInsertRequestSnafu {
448                    reason: format!("cannot add physical column `{}`", source.column_name)
449                }
450            );
451        }
452    }
453    for name in &columns.fields {
454        if !rows.schema.iter().any(|c| c.column_name == *name) {
455            let (column, _) = &columns.by_name[name];
456            ensure!(
457                !column.is_default_impure(),
458                InvalidInsertRequestSnafu {
459                    reason: format!("impure default for physical field `{}`", column.name)
460                }
461            );
462            ensure!(
463                column
464                    .create_default()
465                    .map_err(|e| invalid(e.to_string()))?
466                    .is_some(),
467                InvalidInsertRequestSnafu {
468                    reason: format!("missing required physical field `{}`", column.name)
469                }
470            );
471        }
472    }
473    Ok(())
474}
475
476/// Schema decisions are shared; row values are always checked independently.
477struct RowValidationPlan {
478    input_columns: usize,
479    schema: Vec<ColumnSchema>,
480    nullable: Vec<bool>,
481    defaults: Vec<Value>,
482    timestamp: Option<(usize, TimeUnit, TimeUnit)>,
483}
484
485impl RowValidationPlan {
486    fn new(
487        input_columns: usize,
488        source_timestamp: Option<(usize, TimeUnit)>,
489        mut template: Rows,
490        columns: &[(DataColumn, SemanticType)],
491    ) -> Result<Self> {
492        let by_name: AHashMap<_, _> = columns
493            .iter()
494            .map(|(column, semantic)| (column.name.as_str(), (column, *semantic)))
495            .collect();
496        let mut names = HashSet::new();
497        let mut nullable = Vec::with_capacity(columns.len());
498        for source in &template.schema {
499            ensure!(
500                names.insert(source.column_name.as_str()),
501                InvalidInsertRequestSnafu {
502                    reason: format!("duplicate column `{}`", source.column_name),
503                }
504            );
505            let (target, semantic) = by_name
506                .get(source.column_name.as_str())
507                .ok_or_else(|| invalid(format!("unknown column `{}`", source.column_name)))?;
508            check_definition(source, target, *semantic)?;
509            nullable.push(target.is_nullable());
510        }
511        let missing = columns
512            .iter()
513            .filter(|(column, _)| !names.contains(column.name.as_str()))
514            .collect::<Vec<_>>();
515        let mut row = template.rows.pop().context(InvalidInsertRequestSnafu {
516            reason: "missing normalization template",
517        })?;
518        // Resolve omissions before schemas are merged; an explicit null never uses a default.
519        for (column, semantic) in missing {
520            let value = column
521                .create_default()
522                .map_err(|err| invalid(err.to_string()))?
523                .ok_or_else(|| invalid(format!("missing required column `{}`", column.name)))?;
524            let (datatype, datatype_extension) =
525                ColumnDataTypeWrapper::try_from(column.data_type.clone())
526                    .map_err(|err| invalid(err.to_string()))?
527                    .into_parts();
528            template.schema.push(ColumnSchema {
529                column_name: column.name.clone(),
530                datatype: datatype as i32,
531                semantic_type: *semantic as i32,
532                datatype_extension,
533                options: None,
534            });
535            row.values.push(to_grpc_value(value));
536            nullable.push(column.is_nullable());
537        }
538        ensure!(
539            row.values.len() + input_columns == template.schema.len(),
540            InvalidInsertRequestSnafu {
541                reason: "invalid normalization template"
542            }
543        );
544        let defaults = row.values;
545        for ((source, nullable), value) in template.schema[input_columns..]
546            .iter()
547            .zip(&nullable[input_columns..])
548            .zip(&defaults)
549        {
550            validate_value(value, source, *nullable)?;
551        }
552        let timestamp = source_timestamp.and_then(|(index, source)| {
553            let target = ColumnDataType::try_from(template.schema[index].datatype)
554                .ok()
555                .and_then(api::helper::timestamp_unit)?;
556            (source != target).then_some((index, source, target))
557        });
558        Ok(Self {
559            input_columns,
560            schema: template.schema,
561            nullable,
562            defaults,
563            timestamp,
564        })
565    }
566
567    fn apply_row(&self, row: &mut Row) -> Result<()> {
568        ensure!(
569            row.values.len() == self.input_columns,
570            InvalidInsertRequestSnafu {
571                reason: "column count mismatch",
572            }
573        );
574        if let Some((index, source, target)) = self.timestamp
575            && let Some(value) = row.values[index].value_data.take()
576        {
577            row.values[index].value_data =
578                convert_timestamp_value_data(value, source, target, index)?;
579        }
580        for ((source, nullable), value) in self.schema.iter().zip(&self.nullable).zip(&row.values) {
581            validate_value(value, source, *nullable)?;
582        }
583        row.values.extend_from_slice(&self.defaults);
584        Ok(())
585    }
586}
587
588fn timestamp_column(schema: &[ColumnSchema]) -> Option<(usize, TimeUnit)> {
589    let index = schema
590        .iter()
591        .position(|column| column.semantic_type == SemanticType::Timestamp as i32)?;
592    let unit = ColumnDataType::try_from(schema[index].datatype)
593        .ok()
594        .and_then(api::helper::timestamp_unit)?;
595    Some((index, unit))
596}
597
598fn validate_value(value: &Value, source: &ColumnSchema, nullable: bool) -> Result<()> {
599    ensure!(
600        value_matches_type(value, source.datatype, source.datatype_extension.as_ref()),
601        InvalidInsertRequestSnafu {
602            reason: format!("value datatype differs for `{}`", source.column_name)
603        }
604    );
605    ensure!(
606        value.value_data.is_some() || nullable,
607        InvalidInsertRequestSnafu {
608            reason: format!(
609                "null supplied for non-nullable column `{}`",
610                source.column_name
611            ),
612        }
613    );
614    Ok(())
615}
616
617#[cfg(test)]
618fn validate_and_fill_rows(rows: &mut Rows, columns: &[(DataColumn, SemanticType)]) -> Result<()> {
619    let template = Rows {
620        schema: rows.schema.clone(),
621        rows: vec![Row::default()],
622    };
623    let plan = RowValidationPlan::new(
624        rows.schema.len(),
625        timestamp_column(&rows.schema),
626        template,
627        columns,
628    )?;
629    for row in &mut rows.rows {
630        plan.apply_row(row)?;
631    }
632    rows.schema = plan.schema;
633    Ok(())
634}
635
636#[cfg(test)]
637mod tests {
638    use api::v1::value::ValueData;
639    use api::v1::{ColumnDataType, Row};
640    use datatypes::prelude::ConcreteDataType;
641    use datatypes::schema::ColumnDefaultConstraint;
642    use datatypes::value::Value as DataValue;
643
644    use super::*;
645
646    #[test]
647    fn test_reused_plan_checks_each_value_and_timestamp() {
648        let schema = vec![api::v1::helper::time_index_column_schema(
649            greptime_timestamp(),
650            ColumnDataType::TimestampSecond,
651        )];
652        let mut template = Rows {
653            schema: schema.clone(),
654            rows: Vec::new(),
655        };
656        convert_rows_time_unit(&mut template, TimeUnit::Nanosecond).unwrap();
657        let columns = vec![(
658            DataColumn::new(
659                greptime_timestamp(),
660                ConcreteDataType::timestamp_nanosecond_datatype(),
661                false,
662            )
663            .with_time_index(true),
664            SemanticType::Timestamp,
665        )];
666        template.rows.push(Row::default());
667        let plan =
668            RowValidationPlan::new(schema.len(), timestamp_column(&schema), template, &columns)
669                .unwrap();
670        for (value, valid) in [
671            (Some(ValueData::TimestampSecondValue(1)), true),
672            (Some(ValueData::StringValue("invalid".into())), false),
673            (None, false),
674            (Some(ValueData::TimestampSecondValue(i64::MAX)), false),
675            (Some(ValueData::TimestampSecondValue(2)), true),
676        ] {
677            let mut rows = Rows {
678                schema: schema.clone(),
679                rows: vec![Row {
680                    values: vec![Value { value_data: value }],
681                }],
682            };
683            assert_eq!(plan.apply_row(&mut rows.rows[0]).is_ok(), valid);
684            if valid {
685                assert_eq!(
686                    plan.schema[0].datatype,
687                    ColumnDataType::TimestampNanosecond as i32
688                );
689                assert!(matches!(
690                    rows.rows[0].values[0].value_data,
691                    Some(ValueData::TimestampNanosecondValue(
692                        1_000_000_000 | 2_000_000_000
693                    ))
694                ));
695            }
696        }
697    }
698
699    #[test]
700    fn test_preflight_omissions_defaults_and_explicit_nulls() {
701        let mut columns = default_physical_columns();
702        columns.push((
703            DataColumn::new("host", ConcreteDataType::string_datatype(), false)
704                .with_default_constraint(Some(ColumnDefaultConstraint::Value(DataValue::String(
705                    "unknown".into(),
706                ))))
707                .unwrap(),
708            SemanticType::Tag,
709        ));
710        let row = || Rows {
711            schema: vec![ColumnSchema {
712                column_name: greptime_timestamp().into(),
713                datatype: ColumnDataType::TimestampMillisecond as i32,
714                semantic_type: SemanticType::Timestamp as i32,
715                ..Default::default()
716            }],
717            rows: vec![Row {
718                values: vec![Value {
719                    value_data: Some(ValueData::TimestampMillisecondValue(1)),
720                }],
721            }],
722        };
723        let mut omitted = row();
724        validate_and_fill_rows(&mut omitted, &columns).unwrap();
725        assert_eq!(omitted.schema[2].column_name, "host");
726        assert_eq!(
727            omitted.rows[0].values[2].value_data,
728            Some(ValueData::StringValue("unknown".into()))
729        );
730        omitted.rows[0].values[2] = Value::default();
731        assert!(validate_and_fill_rows(&mut omitted, &columns).is_err());
732        columns[2].0 = DataColumn::new("host", ConcreteDataType::string_datatype(), false);
733        assert!(validate_and_fill_rows(&mut row(), &columns).is_err());
734        columns[2].0.set_nullable();
735        let mut nullable = row();
736        validate_and_fill_rows(&mut nullable, &columns).unwrap();
737        assert_eq!(nullable.rows[0].values[2].value_data, None);
738    }
739
740    #[test]
741    fn test_preflight_existing_columns_and_physical_definitions() {
742        let columns = default_physical_columns();
743        let mut rows = Rows {
744            schema: vec![ColumnSchema {
745                column_name: greptime_timestamp().into(),
746                datatype: ColumnDataType::TimestampMillisecond as i32,
747                semantic_type: SemanticType::Timestamp as i32,
748                ..Default::default()
749            }],
750            rows: vec![Row {
751                values: vec![Value {
752                    value_data: Some(ValueData::TimestampMillisecondValue(1)),
753                }],
754            }],
755        };
756        validate_and_fill_rows(&mut rows, &columns).unwrap();
757        for case in 0..3 {
758            let mut invalid = rows.clone();
759            match case {
760                0 => invalid.schema[0].semantic_type = SemanticType::Tag as i32,
761                1 => invalid.schema[0].datatype = ColumnDataType::TimestampNanosecond as i32,
762                _ => {
763                    invalid.rows[0].values[0].value_data =
764                        Some(ValueData::StringValue("bad".into()))
765                }
766            }
767            assert!(validate_and_fill_rows(&mut invalid, &columns).is_err());
768        }
769        let columns = PhysicalColumns::new(columns);
770        rows.schema[1].column_name = "new_field".into();
771        assert!(validate_physical_columns(&rows, &columns, true).is_err());
772        rows.schema[1].semantic_type = SemanticType::Tag as i32;
773        rows.schema[1].datatype = ColumnDataType::String as i32;
774        assert!(validate_physical_columns(&rows, &columns, false).is_err());
775        assert!(validate_physical_columns(&rows, &columns, true).is_ok());
776    }
777
778    #[test]
779    fn test_preflight_shared_physical_field_defaults() {
780        let rows = Rows {
781            schema: vec![api::v1::helper::time_index_column_schema(
782                greptime_timestamp(),
783                ColumnDataType::TimestampMillisecond,
784            )],
785            ..Default::default()
786        };
787        let required = DataColumn::new("required", ConcreteDataType::float64_datatype(), false);
788        let defaulted = required
789            .clone()
790            .with_default_constraint(Some(ColumnDefaultConstraint::Value(DataValue::Float64(
791                1.0.into(),
792            ))))
793            .unwrap();
794        let impure = DataColumn::new(
795            "impure",
796            ConcreteDataType::timestamp_millisecond_datatype(),
797            false,
798        )
799        .with_default_constraint(Some(ColumnDefaultConstraint::Function(
800            "current_timestamp()".into(),
801        )))
802        .unwrap();
803        for (field, error) in [
804            (required, Some("missing required physical field `required`")),
805            (defaulted, None),
806            (impure, Some("impure default for physical field `impure`")),
807        ] {
808            let mut columns = default_physical_columns();
809            columns.extend((0..1024).map(|index| {
810                (
811                    DataColumn::new(
812                        format!("unrelated_tag_{index}"),
813                        ConcreteDataType::string_datatype(),
814                        true,
815                    ),
816                    SemanticType::Tag,
817                )
818            }));
819            columns.push((field, SemanticType::Field));
820            let columns = PhysicalColumns::new(columns);
821            for auto_create in [false, true] {
822                let result = validate_physical_columns(&rows, &columns, auto_create);
823                if let Some(error) = error {
824                    assert!(result.unwrap_err().to_string().contains(error));
825                } else {
826                    result.unwrap();
827                }
828            }
829        }
830    }
831}
operator/insert/otlp.rs

operator/insert/
otlp.rs
