pipeline/
error.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
15use std::any::Any;
16
17use common_error::ext::ErrorExt;
18use common_error::status_code::StatusCode;
19use common_macro::stack_trace_debug;
20use datatypes::timestamp::TimestampNanosecond;
21use snafu::{Location, Snafu};
22
23#[derive(Snafu)]
24#[snafu(visibility(pub))]
25#[stack_trace_debug]
26pub enum Error {
27    #[snafu(display("Empty input field"))]
28    EmptyInputField {
29        #[snafu(implicit)]
30        location: Location,
31    },
32
33    #[snafu(display("Missing input field"))]
34    MissingInputField {
35        #[snafu(implicit)]
36        location: Location,
37    },
38
39    #[snafu(display(
40        "Field renaming must be a string pair of 'key' and 'rename_to', got: {value:?}"
41    ))]
42    InvalidFieldRename {
43        value: yaml_rust::Yaml,
44        #[snafu(implicit)]
45        location: Location,
46    },
47
48    #[snafu(display("Processor must be a map"))]
49    ProcessorMustBeMap {
50        #[snafu(implicit)]
51        location: Location,
52    },
53
54    #[snafu(display("Processor {processor}: missing field: {field}"))]
55    ProcessorMissingField {
56        processor: String,
57        field: String,
58        #[snafu(implicit)]
59        location: Location,
60    },
61
62    #[snafu(display("Processor {processor}: expect string value, but got {v:?}"))]
63    ProcessorExpectString {
64        processor: String,
65        v: crate::Value,
66        #[snafu(implicit)]
67        location: Location,
68    },
69
70    #[snafu(display("Processor {processor}: unsupported value {val}"))]
71    ProcessorUnsupportedValue {
72        processor: String,
73        val: String,
74        #[snafu(implicit)]
75        location: Location,
76    },
77
78    #[snafu(display("Processor key must be a string"))]
79    ProcessorKeyMustBeString {
80        #[snafu(implicit)]
81        location: Location,
82    },
83
84    #[snafu(display("Processor {kind}: failed to parse {value}"))]
85    ProcessorFailedToParseString {
86        kind: String,
87        value: String,
88        #[snafu(implicit)]
89        location: Location,
90    },
91
92    #[snafu(display("Processor must have a string key"))]
93    ProcessorMustHaveStringKey {
94        #[snafu(implicit)]
95        location: Location,
96    },
97
98    #[snafu(display("Unsupported {processor} processor"))]
99    UnsupportedProcessor {
100        processor: String,
101        #[snafu(implicit)]
102        location: Location,
103    },
104
105    #[snafu(display("Field {field} must be a {ty}"))]
106    FieldMustBeType {
107        field: String,
108        ty: String,
109        #[snafu(implicit)]
110        location: Location,
111    },
112
113    #[snafu(display("Field parse from string failed: {field}"))]
114    FailedParseFieldFromString {
115        #[snafu(source)]
116        error: Box<dyn std::error::Error + Send + Sync>,
117        field: String,
118        #[snafu(implicit)]
119        location: Location,
120    },
121
122    #[snafu(display("Failed to parse {key} as int: {value}"))]
123    FailedToParseIntKey {
124        key: String,
125        value: String,
126        #[snafu(source)]
127        error: std::num::ParseIntError,
128        #[snafu(implicit)]
129        location: Location,
130    },
131
132    #[snafu(display("Failed to parse {value} to int"))]
133    FailedToParseInt {
134        value: String,
135        #[snafu(source)]
136        error: std::num::ParseIntError,
137        #[snafu(implicit)]
138        location: Location,
139    },
140    #[snafu(display("Failed to parse {key} as float: {value}"))]
141    FailedToParseFloatKey {
142        key: String,
143        value: String,
144        #[snafu(source)]
145        error: std::num::ParseFloatError,
146        #[snafu(implicit)]
147        location: Location,
148    },
149
150    #[snafu(display("Processor {kind}: {key} not found in intermediate keys"))]
151    IntermediateKeyIndex {
152        kind: String,
153        key: String,
154        #[snafu(implicit)]
155        location: Location,
156    },
157
158    #[snafu(display("Cmcd {k} missing value in {s}"))]
159    CmcdMissingValue {
160        k: String,
161        s: String,
162        #[snafu(implicit)]
163        location: Location,
164    },
165    #[snafu(display("Part: {part} missing key in {s}"))]
166    CmcdMissingKey {
167        part: String,
168        s: String,
169        #[snafu(implicit)]
170        location: Location,
171    },
172    #[snafu(display("Key must be a string, but got {k:?}"))]
173    KeyMustBeString {
174        k: yaml_rust::Yaml,
175        #[snafu(implicit)]
176        location: Location,
177    },
178
179    #[snafu(display("Csv read error"))]
180    CsvRead {
181        #[snafu(implicit)]
182        location: Location,
183        #[snafu(source)]
184        error: csv::Error,
185    },
186    #[snafu(display("Expected at least one record from csv format, but got none"))]
187    CsvNoRecord {
188        #[snafu(implicit)]
189        location: Location,
190    },
191
192    #[snafu(display("Separator '{separator}' must be a single character, but got '{value}'"))]
193    CsvSeparatorName {
194        separator: String,
195        value: String,
196        #[snafu(implicit)]
197        location: Location,
198    },
199
200    #[snafu(display("Quote '{quote}' must be a single character, but got '{value}'"))]
201    CsvQuoteName {
202        quote: String,
203        value: String,
204        #[snafu(implicit)]
205        location: Location,
206    },
207
208    #[snafu(display("Parse date timezone error {value}"))]
209    DateParseTimezone {
210        value: String,
211        #[snafu(source)]
212        error: chrono_tz::ParseError,
213        #[snafu(implicit)]
214        location: Location,
215    },
216
217    #[snafu(display("Parse date error {value}"))]
218    DateParse {
219        value: String,
220        #[snafu(source)]
221        error: chrono::ParseError,
222        #[snafu(implicit)]
223        location: Location,
224    },
225
226    #[snafu(display("Failed to get local timezone"))]
227    DateFailedToGetLocalTimezone {
228        #[snafu(implicit)]
229        location: Location,
230    },
231
232    #[snafu(display("Failed to get timestamp"))]
233    DateFailedToGetTimestamp {
234        #[snafu(implicit)]
235        location: Location,
236    },
237
238    #[snafu(display("Invalid Pattern: '{s}'. {detail}"))]
239    DissectInvalidPattern {
240        s: String,
241        detail: String,
242        #[snafu(implicit)]
243        location: Location,
244    },
245
246    #[snafu(display("Empty pattern is not allowed"))]
247    DissectEmptyPattern {
248        #[snafu(implicit)]
249        location: Location,
250    },
251    #[snafu(display("Split: '{split}' exceeds the input"))]
252    DissectSplitExceedsInput {
253        split: String,
254        #[snafu(implicit)]
255        location: Location,
256    },
257    #[snafu(display("Split: '{split}' does not match the input '{input}'"))]
258    DissectSplitNotMatchInput {
259        split: String,
260        input: String,
261        #[snafu(implicit)]
262        location: Location,
263    },
264    #[snafu(display("Consecutive names are not allowed: '{name1}' '{name2}'"))]
265    DissectConsecutiveNames {
266        name1: String,
267        name2: String,
268        #[snafu(implicit)]
269        location: Location,
270    },
271    #[snafu(display("No matching pattern found"))]
272    DissectNoMatchingPattern {
273        #[snafu(implicit)]
274        location: Location,
275    },
276    #[snafu(display("Modifier '{m}' already set, but found {modifier}"))]
277    DissectModifierAlreadySet {
278        m: String,
279        modifier: String,
280        #[snafu(implicit)]
281        location: Location,
282    },
283
284    #[snafu(display("Append Order modifier is already set to '{n}', cannot be set to {order}"))]
285    DissectAppendOrderAlreadySet {
286        n: String,
287        order: u32,
288        #[snafu(implicit)]
289        location: Location,
290    },
291    #[snafu(display("Order can only be set to Append Modifier, current modifier is {m}"))]
292    DissectOrderOnlyAppend {
293        m: String,
294        #[snafu(implicit)]
295        location: Location,
296    },
297
298    #[snafu(display("Order can only be set to Append Modifier"))]
299    DissectOrderOnlyAppendModifier {
300        #[snafu(implicit)]
301        location: Location,
302    },
303
304    #[snafu(display("End modifier already set: '{m}'"))]
305    DissectEndModifierAlreadySet {
306        m: String,
307        #[snafu(implicit)]
308        location: Location,
309    },
310    #[snafu(display("Invalid resolution: {resolution}"))]
311    EpochInvalidResolution {
312        resolution: String,
313        #[snafu(implicit)]
314        location: Location,
315    },
316    #[snafu(display("Pattern is required"))]
317    GsubPatternRequired {
318        #[snafu(implicit)]
319        location: Location,
320    },
321    #[snafu(display("Replacement is required"))]
322    GsubReplacementRequired {
323        #[snafu(implicit)]
324        location: Location,
325    },
326    #[snafu(display("Invalid regex pattern: {pattern}"))]
327    Regex {
328        #[snafu(source)]
329        error: regex::Error,
330        pattern: String,
331        #[snafu(implicit)]
332        location: Location,
333    },
334    #[snafu(display("Separator is required"))]
335    JoinSeparatorRequired {
336        #[snafu(implicit)]
337        location: Location,
338    },
339    #[snafu(display("Invalid method: {method}"))]
340    LetterInvalidMethod {
341        method: String,
342        #[snafu(implicit)]
343        location: Location,
344    },
345    #[snafu(display("No named group found in regex {origin}"))]
346    RegexNamedGroupNotFound {
347        origin: String,
348        #[snafu(implicit)]
349        location: Location,
350    },
351    #[snafu(display("No valid field found in {processor} processor"))]
352    RegexNoValidField {
353        processor: String,
354        #[snafu(implicit)]
355        location: Location,
356    },
357    #[snafu(display("No valid pattern found in {processor} processor"))]
358    RegexNoValidPattern {
359        processor: String,
360        #[snafu(implicit)]
361        location: Location,
362    },
363    #[snafu(display("Invalid method: {s}"))]
364    UrlEncodingInvalidMethod {
365        s: String,
366        #[snafu(implicit)]
367        location: Location,
368    },
369    #[snafu(display("Wrong digest pattern: {pattern}"))]
370    DigestPatternInvalid {
371        pattern: String,
372        #[snafu(implicit)]
373        location: Location,
374    },
375    #[snafu(display("Url decoding error"))]
376    UrlEncodingDecode {
377        #[snafu(source)]
378        error: std::string::FromUtf8Error,
379        #[snafu(implicit)]
380        location: Location,
381    },
382    #[snafu(display("Invalid transform on_failure value: {value}"))]
383    TransformOnFailureInvalidValue {
384        value: String,
385        #[snafu(implicit)]
386        location: Location,
387    },
388    #[snafu(display("Transform element must be a map"))]
389    TransformElementMustBeMap {
390        #[snafu(implicit)]
391        location: Location,
392    },
393    #[snafu(display("Transform fields must be set."))]
394    TransformFieldMustBeSet {
395        #[snafu(implicit)]
396        location: Location,
397    },
398    #[snafu(display("Transform {fields:?} type MUST BE set."))]
399    TransformTypeMustBeSet {
400        fields: String,
401        #[snafu(implicit)]
402        location: Location,
403    },
404    #[snafu(display("Column name must be unique, but got duplicated: {duplicates}"))]
405    TransformColumnNameMustBeUnique {
406        duplicates: String,
407        #[snafu(implicit)]
408        location: Location,
409    },
410    #[snafu(display(
411        "Illegal to set multiple timestamp Index columns, please set only one: {columns}"
412    ))]
413    TransformMultipleTimestampIndex {
414        columns: String,
415        #[snafu(implicit)]
416        location: Location,
417    },
418    #[snafu(display("Transform must have exactly one field specified as timestamp Index, but got {count}: {columns}"))]
419    TransformTimestampIndexCount {
420        count: usize,
421        columns: String,
422        #[snafu(implicit)]
423        location: Location,
424    },
425    #[snafu(display("Exactly one time-related processor and one timestamp value is required to use auto transform. `ignore_missing` can not be set to true."))]
426    AutoTransformOneTimestamp {
427        #[snafu(implicit)]
428        location: Location,
429    },
430    #[snafu(display("Invalid Pipeline doc version number: {}", version))]
431    InvalidVersionNumber {
432        version: String,
433        #[snafu(implicit)]
434        location: Location,
435    },
436    #[snafu(display("Null type not supported"))]
437    CoerceUnsupportedNullType {
438        #[snafu(implicit)]
439        location: Location,
440    },
441    #[snafu(display("Null type not supported when to coerce '{ty}' type"))]
442    CoerceUnsupportedNullTypeTo {
443        ty: String,
444        #[snafu(implicit)]
445        location: Location,
446    },
447    #[snafu(display("Type: {ty} value not supported for Epoch"))]
448    CoerceUnsupportedEpochType {
449        ty: String,
450        #[snafu(implicit)]
451        location: Location,
452    },
453    #[snafu(display("Failed to coerce string value '{s}' to type '{ty}'"))]
454    CoerceStringToType {
455        s: String,
456        ty: String,
457        #[snafu(implicit)]
458        location: Location,
459    },
460    #[snafu(display("Can not coerce json type to {ty}"))]
461    CoerceJsonTypeTo {
462        ty: String,
463        #[snafu(implicit)]
464        location: Location,
465    },
466    #[snafu(display(
467        "Can not coerce {ty} to json type. we only consider object and array to be json types."
468    ))]
469    CoerceTypeToJson {
470        ty: String,
471        #[snafu(implicit)]
472        location: Location,
473    },
474    #[snafu(display("Failed to coerce value: {msg}"))]
475    CoerceIncompatibleTypes {
476        msg: String,
477        #[snafu(implicit)]
478        location: Location,
479    },
480    #[snafu(display(
481        "Invalid resolution: '{resolution}'. Available resolutions: {valid_resolution}"
482    ))]
483    ValueInvalidResolution {
484        resolution: String,
485        valid_resolution: String,
486        #[snafu(implicit)]
487        location: Location,
488    },
489
490    #[snafu(display("Failed to parse type: '{t}'"))]
491    ValueParseType {
492        t: String,
493        #[snafu(implicit)]
494        location: Location,
495    },
496
497    #[snafu(display("Failed to parse {ty}: {v}"))]
498    ValueParseInt {
499        ty: String,
500        v: String,
501        #[snafu(source)]
502        error: std::num::ParseIntError,
503        #[snafu(implicit)]
504        location: Location,
505    },
506
507    #[snafu(display("Failed to parse {ty}: {v}"))]
508    ValueParseFloat {
509        ty: String,
510        v: String,
511        #[snafu(source)]
512        error: std::num::ParseFloatError,
513        #[snafu(implicit)]
514        location: Location,
515    },
516
517    #[snafu(display("Failed to parse {ty}: {v}"))]
518    ValueParseBoolean {
519        ty: String,
520        v: String,
521        #[snafu(source)]
522        error: std::str::ParseBoolError,
523        #[snafu(implicit)]
524        location: Location,
525    },
526    #[snafu(display("Default value not unsupported for type {value}"))]
527    ValueDefaultValueUnsupported {
528        value: String,
529        #[snafu(implicit)]
530        location: Location,
531    },
532
533    #[snafu(display("Unsupported yaml type: {value:?}"))]
534    ValueUnsupportedYamlType {
535        value: yaml_rust::Yaml,
536        #[snafu(implicit)]
537        location: Location,
538    },
539
540    #[snafu(display("key in Hash must be a string, but got {value:?}"))]
541    ValueYamlKeyMustBeString {
542        value: yaml_rust::Yaml,
543        #[snafu(implicit)]
544        location: Location,
545    },
546
547    #[snafu(display("Yaml load error."))]
548    YamlLoad {
549        #[snafu(source)]
550        error: yaml_rust::ScanError,
551        #[snafu(implicit)]
552        location: Location,
553    },
554    #[snafu(display("Yaml parse error."))]
555    YamlParse {
556        #[snafu(implicit)]
557        location: Location,
558    },
559    #[snafu(display("Input value must be an object"))]
560    InputValueMustBeObject {
561        #[snafu(implicit)]
562        location: Location,
563    },
564
565    #[snafu(display("Column options error"))]
566    ColumnOptions {
567        #[snafu(source)]
568        source: api::error::Error,
569        #[snafu(implicit)]
570        location: Location,
571    },
572    #[snafu(display("Unsupported index type: {value}"))]
573    UnsupportedIndexType {
574        value: String,
575        #[snafu(implicit)]
576        location: Location,
577    },
578    #[snafu(display("Unsupported number type: {value:?}"))]
579    UnsupportedNumberType {
580        value: serde_json::Number,
581        #[snafu(implicit)]
582        location: Location,
583    },
584    #[snafu(display("Failed to parse json"))]
585    JsonParse {
586        #[snafu(source)]
587        error: serde_json::Error,
588        #[snafu(implicit)]
589        location: Location,
590    },
591    #[snafu(display("Column datatype mismatch. For column: {column}, expected datatype: {expected}, actual datatype: {actual}"))]
592    IdentifyPipelineColumnTypeMismatch {
593        column: String,
594        expected: String,
595        actual: String,
596        #[snafu(implicit)]
597        location: Location,
598    },
599    #[snafu(display("Parse json path error"))]
600    JsonPathParse {
601        #[snafu(implicit)]
602        location: Location,
603        #[snafu(source)]
604        error: jsonpath_rust::JsonPathParserError,
605    },
606    #[snafu(display("Json path result index not number"))]
607    JsonPathParseResultIndex {
608        #[snafu(implicit)]
609        location: Location,
610    },
611    #[snafu(display("Field is required for dispatcher"))]
612    FieldRequiredForDispatcher,
613    #[snafu(display("Table_suffix is required for dispatcher rule"))]
614    TableSuffixRequiredForDispatcherRule,
615    #[snafu(display("Value is required for dispatcher rule"))]
616    ValueRequiredForDispatcherRule,
617    #[snafu(display(
618        "Reached max nested levels when flattening JSON object: {max_nested_levels}"
619    ))]
620    ReachedMaxNestedLevels {
621        max_nested_levels: usize,
622        #[snafu(implicit)]
623        location: Location,
624    },
625
626    #[snafu(display("Pipeline table not found"))]
627    PipelineTableNotFound {
628        #[snafu(implicit)]
629        location: Location,
630    },
631
632    #[snafu(display("Failed to insert pipeline to pipelines table"))]
633    InsertPipeline {
634        #[snafu(source)]
635        source: operator::error::Error,
636        #[snafu(implicit)]
637        location: Location,
638    },
639
640    #[snafu(display("Pipeline not found, name: {}, version: {}", name, version.map(|ts| ts.0.to_iso8601_string()).unwrap_or("latest".to_string())))]
641    PipelineNotFound {
642        name: String,
643        version: Option<TimestampNanosecond>,
644        #[snafu(implicit)]
645        location: Location,
646    },
647
648    #[snafu(display(
649        "Multiple pipelines with different schemas found, but none under current schema. Please replicate one of them or delete until only one schema left. schemas: {}",
650        schemas
651    ))]
652    MultiPipelineWithDiffSchema {
653        schemas: String,
654        #[snafu(implicit)]
655        location: Location,
656    },
657
658    #[snafu(display(
659        "The return value's length of the record batch does not match, see debug log for details"
660    ))]
661    RecordBatchLenNotMatch {
662        #[snafu(implicit)]
663        location: Location,
664    },
665
666    #[snafu(display("Failed to collect record batch"))]
667    CollectRecords {
668        #[snafu(implicit)]
669        location: Location,
670        #[snafu(source)]
671        source: common_recordbatch::error::Error,
672    },
673
674    #[snafu(display("A valid table suffix template is required for tablesuffix section"))]
675    RequiredTableSuffixTemplate,
676
677    #[snafu(display("Invalid table suffix template, input: {}", input))]
678    InvalidTableSuffixTemplate {
679        input: String,
680        #[snafu(implicit)]
681        location: Location,
682    },
683
684    #[snafu(display("Failed to compile VRL, {}", msg))]
685    CompileVrl {
686        msg: String,
687        #[snafu(implicit)]
688        location: Location,
689    },
690
691    #[snafu(display("Failed to execute VRL, {}", msg))]
692    ExecuteVrl {
693        msg: String,
694        #[snafu(implicit)]
695        location: Location,
696    },
697
698    #[snafu(display("Float is not a number: {}", input_float))]
699    FloatNaN {
700        input_float: f64,
701        #[snafu(implicit)]
702        location: Location,
703    },
704
705    #[snafu(display("Invalid timestamp value: {}", input))]
706    InvalidTimestamp {
707        input: String,
708        #[snafu(implicit)]
709        location: Location,
710    },
711
712    #[snafu(display("Failed to convert bytes to utf8"))]
713    BytesToUtf8 {
714        #[snafu(source)]
715        error: std::string::FromUtf8Error,
716        #[snafu(implicit)]
717        location: Location,
718    },
719
720    #[snafu(display("Please don't use regex in Vrl script"))]
721    VrlRegexValue {
722        #[snafu(implicit)]
723        location: Location,
724    },
725
726    #[snafu(display("Vrl script should return `.` in the end"))]
727    VrlReturnValue {
728        #[snafu(implicit)]
729        location: Location,
730    },
731
732    #[snafu(display("Failed to cast type, msg: {}", msg))]
733    CastType {
734        msg: String,
735        #[snafu(implicit)]
736        location: Location,
737    },
738
739    #[snafu(display("Top level value must be map"))]
740    ValueMustBeMap {
741        #[snafu(implicit)]
742        location: Location,
743    },
744
745    #[snafu(display("Failed to build DataFusion logical plan"))]
746    BuildDfLogicalPlan {
747        #[snafu(source)]
748        error: datafusion_common::DataFusionError,
749        #[snafu(implicit)]
750        location: Location,
751    },
752
753    #[snafu(display("Failed to execute internal statement"))]
754    ExecuteInternalStatement {
755        #[snafu(source)]
756        source: query::error::Error,
757        #[snafu(implicit)]
758        location: Location,
759    },
760
761    #[snafu(display("Failed to create dataframe"))]
762    DataFrame {
763        #[snafu(source)]
764        source: query::error::Error,
765        #[snafu(implicit)]
766        location: Location,
767    },
768
769    #[snafu(display("General catalog error"))]
770    Catalog {
771        #[snafu(source)]
772        source: catalog::error::Error,
773        #[snafu(implicit)]
774        location: Location,
775    },
776
777    #[snafu(display("Failed to create table"))]
778    CreateTable {
779        #[snafu(source)]
780        source: operator::error::Error,
781        #[snafu(implicit)]
782        location: Location,
783    },
784
785    #[snafu(display("Invalid pipeline version format: {}", version))]
786    InvalidPipelineVersion {
787        version: String,
788        #[snafu(implicit)]
789        location: Location,
790    },
791
792    #[snafu(display("Invalid custom time index config: {}, reason: {}", config, reason))]
793    InvalidCustomTimeIndex {
794        config: String,
795        reason: String,
796        #[snafu(implicit)]
797        location: Location,
798    },
799
800    #[snafu(display("Pipeline is required for this API."))]
801    PipelineMissing {
802        #[snafu(implicit)]
803        location: Location,
804    },
805
806    #[snafu(display("Time index must be non null."))]
807    TimeIndexMustBeNonNull {
808        #[snafu(implicit)]
809        location: Location,
810    },
811}
812
813pub type Result<T> = std::result::Result<T, Error>;
814
815impl ErrorExt for Error {
816    fn status_code(&self) -> StatusCode {
817        use Error::*;
818        match self {
819            CastType { .. } => StatusCode::Unexpected,
820            PipelineTableNotFound { .. } => StatusCode::TableNotFound,
821            InsertPipeline { source, .. } => source.status_code(),
822            CollectRecords { source, .. } => source.status_code(),
823            PipelineNotFound { .. }
824            | InvalidPipelineVersion { .. }
825            | InvalidCustomTimeIndex { .. }
826            | TimeIndexMustBeNonNull { .. } => StatusCode::InvalidArguments,
827            MultiPipelineWithDiffSchema { .. } | ValueMustBeMap { .. } => StatusCode::IllegalState,
828            BuildDfLogicalPlan { .. } | RecordBatchLenNotMatch { .. } => StatusCode::Internal,
829            ExecuteInternalStatement { source, .. } => source.status_code(),
830            DataFrame { source, .. } => source.status_code(),
831            Catalog { source, .. } => source.status_code(),
832            CreateTable { source, .. } => source.status_code(),
833
834            EmptyInputField { .. }
835            | MissingInputField { .. }
836            | InvalidFieldRename { .. }
837            | ProcessorMustBeMap { .. }
838            | ProcessorMissingField { .. }
839            | ProcessorExpectString { .. }
840            | ProcessorUnsupportedValue { .. }
841            | ProcessorKeyMustBeString { .. }
842            | ProcessorFailedToParseString { .. }
843            | ProcessorMustHaveStringKey { .. }
844            | UnsupportedProcessor { .. }
845            | FieldMustBeType { .. }
846            | FailedParseFieldFromString { .. }
847            | FailedToParseIntKey { .. }
848            | FailedToParseInt { .. }
849            | FailedToParseFloatKey { .. }
850            | IntermediateKeyIndex { .. }
851            | CmcdMissingValue { .. }
852            | CmcdMissingKey { .. }
853            | KeyMustBeString { .. }
854            | CsvRead { .. }
855            | CsvNoRecord { .. }
856            | CsvSeparatorName { .. }
857            | CsvQuoteName { .. }
858            | DateParseTimezone { .. }
859            | DateParse { .. }
860            | DateFailedToGetLocalTimezone { .. }
861            | DateFailedToGetTimestamp { .. }
862            | DissectInvalidPattern { .. }
863            | DissectEmptyPattern { .. }
864            | DissectSplitExceedsInput { .. }
865            | DissectSplitNotMatchInput { .. }
866            | DissectConsecutiveNames { .. }
867            | DissectNoMatchingPattern { .. }
868            | DissectModifierAlreadySet { .. }
869            | DissectAppendOrderAlreadySet { .. }
870            | DissectOrderOnlyAppend { .. }
871            | DissectOrderOnlyAppendModifier { .. }
872            | DissectEndModifierAlreadySet { .. }
873            | EpochInvalidResolution { .. }
874            | GsubPatternRequired { .. }
875            | GsubReplacementRequired { .. }
876            | Regex { .. }
877            | JoinSeparatorRequired { .. }
878            | LetterInvalidMethod { .. }
879            | RegexNamedGroupNotFound { .. }
880            | RegexNoValidField { .. }
881            | RegexNoValidPattern { .. }
882            | UrlEncodingInvalidMethod { .. }
883            | DigestPatternInvalid { .. }
884            | UrlEncodingDecode { .. }
885            | TransformOnFailureInvalidValue { .. }
886            | TransformElementMustBeMap { .. }
887            | TransformFieldMustBeSet { .. }
888            | TransformTypeMustBeSet { .. }
889            | TransformColumnNameMustBeUnique { .. }
890            | TransformMultipleTimestampIndex { .. }
891            | TransformTimestampIndexCount { .. }
892            | AutoTransformOneTimestamp { .. }
893            | InvalidVersionNumber { .. }
894            | CoerceUnsupportedNullType { .. }
895            | CoerceUnsupportedNullTypeTo { .. }
896            | CoerceUnsupportedEpochType { .. }
897            | CoerceStringToType { .. }
898            | CoerceJsonTypeTo { .. }
899            | CoerceTypeToJson { .. }
900            | CoerceIncompatibleTypes { .. }
901            | ValueInvalidResolution { .. }
902            | ValueParseType { .. }
903            | ValueParseInt { .. }
904            | ValueParseFloat { .. }
905            | ValueParseBoolean { .. }
906            | ValueDefaultValueUnsupported { .. }
907            | ValueUnsupportedYamlType { .. }
908            | ValueYamlKeyMustBeString { .. }
909            | YamlLoad { .. }
910            | YamlParse { .. }
911            | InputValueMustBeObject { .. }
912            | ColumnOptions { .. }
913            | UnsupportedIndexType { .. }
914            | UnsupportedNumberType { .. }
915            | IdentifyPipelineColumnTypeMismatch { .. }
916            | JsonParse { .. }
917            | JsonPathParse { .. }
918            | JsonPathParseResultIndex { .. }
919            | FieldRequiredForDispatcher
920            | TableSuffixRequiredForDispatcherRule
921            | ValueRequiredForDispatcherRule
922            | ReachedMaxNestedLevels { .. }
923            | RequiredTableSuffixTemplate
924            | InvalidTableSuffixTemplate { .. }
925            | CompileVrl { .. }
926            | ExecuteVrl { .. }
927            | FloatNaN { .. }
928            | BytesToUtf8 { .. }
929            | InvalidTimestamp { .. }
930            | VrlRegexValue { .. }
931            | VrlReturnValue { .. }
932            | PipelineMissing { .. } => StatusCode::InvalidArguments,
933        }
934    }
935
936    fn as_any(&self) -> &dyn Any {
937        self
938    }
939}