1use crate::config::TransformSpec;
7use crate::error::{CliError, CliResult};
8#[cfg(feature = "transforms")]
9use faucet_core::{
10 CastOnError, CastType, HashAlgorithm, HashEncoding, JsonParseOnError, KeyCaseMode,
11 ValueCaseMode,
12};
13#[cfg(any(feature = "transforms", feature = "transform-cdc-unwrap"))]
14use faucet_core::{JsonSchema, schema_for};
15use faucet_core::{RecordTransform, TransformStage};
16#[cfg(any(feature = "transforms", feature = "transform-cdc-unwrap"))]
17use serde::Deserialize;
18use serde_json::Value;
19#[cfg(feature = "transforms")]
20use std::collections::HashMap;
21
22#[cfg(feature = "transforms")]
24#[derive(Debug, Deserialize, JsonSchema)]
25struct FlattenConfig {
26 #[serde(default = "default_separator")]
28 separator: String,
29}
30
31#[cfg(feature = "transforms")]
32fn default_separator() -> String {
33 "__".to_owned()
34}
35
36#[cfg(feature = "transforms")]
38#[derive(Debug, Deserialize, JsonSchema)]
39struct RenameKeysConfig {
40 pattern: String,
42 replacement: String,
44}
45
46#[cfg(feature = "transforms")]
47#[derive(Debug, Deserialize, JsonSchema)]
48struct FieldsConfig {
49 fields: Vec<String>,
51}
52
53#[cfg(feature = "transforms")]
54#[derive(Debug, Deserialize, JsonSchema)]
55struct SetConfig {
56 values: serde_json::Map<String, Value>,
58}
59
60#[cfg(feature = "transforms")]
61#[derive(Debug, Deserialize, JsonSchema)]
62struct RenameFieldConfig {
63 fields: HashMap<String, String>,
65}
66
67#[cfg(feature = "transforms")]
68#[derive(Debug, Deserialize, JsonSchema)]
69struct CastConfig {
70 fields: HashMap<String, CastType>,
72 #[serde(default)]
74 on_error: CastOnError,
75}
76
77#[cfg(feature = "transforms")]
78#[derive(Debug, Deserialize, JsonSchema)]
79struct RedactConfig {
80 fields: Vec<String>,
82 #[serde(default = "default_mask")]
84 mask: Value,
85}
86
87#[cfg(feature = "transforms")]
88fn default_mask() -> Value {
89 Value::String("***".to_owned())
90}
91
92#[cfg(feature = "transforms")]
93#[derive(Debug, Deserialize, JsonSchema)]
94struct ValueCaseConfig {
95 fields: Vec<String>,
97 mode: ValueCaseMode,
99}
100
101#[cfg(feature = "transforms")]
102#[derive(Debug, Deserialize, JsonSchema)]
103struct SpellSymbolsConfig {
104 #[serde(default)]
106 extra: HashMap<String, String>,
107 #[serde(default = "default_spell_separator")]
109 separator: String,
110}
111
112#[cfg(feature = "transforms")]
113fn default_spell_separator() -> String {
114 " ".to_owned()
115}
116
117#[cfg(feature = "transforms")]
118#[derive(Debug, Deserialize, JsonSchema)]
119struct KeysCaseConfig {
120 mode: KeyCaseMode,
122}
123
124#[cfg(feature = "transforms")]
125#[derive(Debug, Deserialize, JsonSchema)]
126struct HashConfig {
127 fields: Vec<String>,
129 #[serde(default)]
131 algorithm: HashAlgorithm,
132 #[serde(default)]
134 encoding: HashEncoding,
135 #[serde(default, skip_serializing_if = "Option::is_none")]
137 salt: Option<String>,
138 #[serde(default, skip_serializing_if = "Option::is_none")]
141 into: Option<String>,
142}
143
144#[cfg(feature = "transforms")]
145#[derive(Debug, Deserialize, JsonSchema)]
146struct JsonParseConfig {
147 fields: Vec<String>,
149 #[serde(default)]
151 on_error: JsonParseOnError,
152 #[serde(default, skip_serializing_if = "Option::is_none")]
155 into: Option<String>,
156}
157
158#[cfg(feature = "transforms")]
159#[derive(Debug, Deserialize, JsonSchema)]
160struct CoalesceConfig {
161 field: String,
163 #[serde(default, skip_serializing_if = "Option::is_none")]
165 default: Option<Value>,
166 #[serde(default)]
168 from: Vec<String>,
169 #[serde(default)]
171 treat_empty_string_as_null: bool,
172}
173
174#[cfg(feature = "transforms")]
175#[derive(Debug, Deserialize, JsonSchema)]
176struct SplitConfig {
177 field: String,
179 delimiter: String,
181 #[serde(default)]
183 trim: bool,
184 #[serde(default, skip_serializing_if = "Option::is_none")]
186 into: Option<String>,
187}
188
189#[cfg(feature = "transforms")]
190#[derive(Debug, Deserialize, JsonSchema)]
191struct JoinConfig {
192 field: String,
194 delimiter: String,
196 #[serde(default, skip_serializing_if = "Option::is_none")]
198 into: Option<String>,
199}
200
201#[cfg(feature = "transform-filter")]
202#[derive(Debug, Deserialize, JsonSchema)]
203struct FilterConfig {
204 path: String,
206 op: faucet_core::FilterOp,
208 #[serde(default, skip_serializing_if = "Option::is_none")]
210 value: Option<Value>,
211}
212
213#[cfg(feature = "transform-explode")]
214#[derive(Debug, Deserialize, JsonSchema)]
215struct ExplodeConfig {
216 path: String,
218 #[serde(default, skip_serializing_if = "Option::is_none")]
221 prefix: Option<String>,
222 #[serde(default = "default_explode_separator_cli")]
224 separator: String,
225 #[serde(default)]
228 on_missing: faucet_core::OnMissing,
229}
230
231#[cfg(feature = "transform-explode")]
232fn default_explode_separator_cli() -> String {
233 "_".to_owned()
234}
235
236#[cfg(feature = "transform-cdc-unwrap")]
237#[derive(Debug, Deserialize, JsonSchema)]
238struct CdcUnwrapConfig {
239 #[serde(default = "cdc_op_field")]
241 op_field: String,
242 #[serde(default = "cdc_after_field")]
244 after_field: String,
245 #[serde(default = "cdc_before_field")]
247 before_field: String,
248 #[serde(default = "cdc_key_field")]
250 key_field: String,
251 #[serde(default = "cdc_marker_field")]
253 marker_field: String,
254 #[serde(default = "cdc_delete_ops")]
256 delete_ops: Vec<String>,
257 #[serde(default = "cdc_drop_ops")]
259 drop_ops: Vec<String>,
260}
261
262#[cfg(feature = "transform-cdc-unwrap")]
263fn cdc_op_field() -> String {
264 "op".into()
265}
266#[cfg(feature = "transform-cdc-unwrap")]
267fn cdc_after_field() -> String {
268 "after".into()
269}
270#[cfg(feature = "transform-cdc-unwrap")]
271fn cdc_before_field() -> String {
272 "before".into()
273}
274#[cfg(feature = "transform-cdc-unwrap")]
275fn cdc_key_field() -> String {
276 "document_key".into()
277}
278#[cfg(feature = "transform-cdc-unwrap")]
279fn cdc_marker_field() -> String {
280 "__op".into()
281}
282#[cfg(feature = "transform-cdc-unwrap")]
283fn cdc_delete_ops() -> Vec<String> {
284 vec!["d".into(), "delete".into()]
285}
286#[cfg(feature = "transform-cdc-unwrap")]
287fn cdc_drop_ops() -> Vec<String> {
288 vec!["ddl".into(), "truncate".into()]
289}
290
291struct TransformDef {
298 kind: &'static str,
299 description: &'static str,
300 schema_fn: fn() -> Value,
301 compile_fn: fn(&str, Value) -> CliResult<TransformStage>,
302}
303
304fn registry() -> Vec<TransformDef> {
310 #[allow(unused_mut)]
311 let mut defs: Vec<TransformDef> = Vec::new();
312 #[cfg(feature = "transforms")]
313 {
314 defs.extend(vec![
315 TransformDef {
316 kind: "flatten",
317 description: "Flatten nested objects into a single level (configurable separator).",
318 schema_fn: || schema::<FlattenConfig>(),
319 compile_fn: |kind, config| {
320 let cfg = decode::<FlattenConfig>(kind, config)?;
321 Ok(TransformStage::Map(RecordTransform::Flatten {
322 separator: cfg.separator,
323 }))
324 },
325 },
326 TransformDef {
327 kind: "rename_keys",
328 description: "Rewrite every key via a regex pattern + replacement.",
329 schema_fn: || schema::<RenameKeysConfig>(),
330 compile_fn: |kind, config| {
331 let cfg = decode::<RenameKeysConfig>(kind, config)?;
332 Ok(TransformStage::Map(RecordTransform::RenameKeys {
333 pattern: cfg.pattern,
334 replacement: cfg.replacement,
335 }))
336 },
337 },
338 TransformDef {
339 kind: "keys_case",
340 description: "Re-case every key (snake / camel / pascal / kebab / screaming_snake).",
341 schema_fn: || schema::<KeysCaseConfig>(),
342 compile_fn: |kind, config| {
343 let cfg = decode::<KeysCaseConfig>(kind, config)?;
344 Ok(TransformStage::Map(RecordTransform::KeysCase {
345 mode: cfg.mode,
346 }))
347 },
348 },
349 TransformDef {
350 kind: "select",
351 description: "Keep only the listed top-level fields; drop the rest.",
352 schema_fn: || schema::<FieldsConfig>(),
353 compile_fn: |kind, config| {
354 let cfg = decode::<FieldsConfig>(kind, config)?;
355 Ok(TransformStage::Map(RecordTransform::Select {
356 fields: cfg.fields,
357 }))
358 },
359 },
360 TransformDef {
361 kind: "drop",
362 description: "Remove the listed top-level fields.",
363 schema_fn: || schema::<FieldsConfig>(),
364 compile_fn: |kind, config| {
365 let cfg = decode::<FieldsConfig>(kind, config)?;
366 Ok(TransformStage::Map(RecordTransform::Drop {
367 fields: cfg.fields,
368 }))
369 },
370 },
371 TransformDef {
372 kind: "set",
373 description: "Set named fields to constant values on every record.",
374 schema_fn: || schema::<SetConfig>(),
375 compile_fn: |kind, config| {
376 let cfg = decode::<SetConfig>(kind, config)?;
377 Ok(TransformStage::Map(RecordTransform::Set {
378 values: cfg.values,
379 }))
380 },
381 },
382 TransformDef {
383 kind: "rename_field",
384 description: "Rename specific top-level fields by name.",
385 schema_fn: || schema::<RenameFieldConfig>(),
386 compile_fn: |kind, config| {
387 let cfg = decode::<RenameFieldConfig>(kind, config)?;
388 Ok(TransformStage::Map(RecordTransform::RenameField {
389 fields: cfg.fields,
390 }))
391 },
392 },
393 TransformDef {
394 kind: "cast",
395 description: "Coerce named fields to int / float / bool / string / timestamp.",
396 schema_fn: || schema::<CastConfig>(),
397 compile_fn: |kind, config| {
398 let cfg = decode::<CastConfig>(kind, config)?;
399 Ok(TransformStage::Map(RecordTransform::Cast {
400 fields: cfg.fields,
401 on_error: cfg.on_error,
402 }))
403 },
404 },
405 TransformDef {
406 kind: "redact",
407 description: "Overwrite the listed fields with a mask value (default `***`).",
408 schema_fn: || schema::<RedactConfig>(),
409 compile_fn: |kind, config| {
410 let cfg = decode::<RedactConfig>(kind, config)?;
411 Ok(TransformStage::Map(RecordTransform::Redact {
412 fields: cfg.fields,
413 mask: cfg.mask,
414 }))
415 },
416 },
417 TransformDef {
418 kind: "value_case",
419 description: "Lowercase, uppercase, or trim the value of named string fields.",
420 schema_fn: || schema::<ValueCaseConfig>(),
421 compile_fn: |kind, config| {
422 let cfg = decode::<ValueCaseConfig>(kind, config)?;
423 Ok(TransformStage::Map(RecordTransform::ValueCase {
424 fields: cfg.fields,
425 mode: cfg.mode,
426 }))
427 },
428 },
429 TransformDef {
430 kind: "spell_symbols",
431 description: "Replace punctuation/symbols in string values with their spelled-out words.",
432 schema_fn: || schema::<SpellSymbolsConfig>(),
433 compile_fn: |kind, config| {
434 let cfg = decode::<SpellSymbolsConfig>(kind, config)?;
435 Ok(TransformStage::Map(RecordTransform::SpellSymbols {
436 extra: cfg.extra,
437 separator: cfg.separator,
438 }))
439 },
440 },
441 TransformDef {
442 kind: "hash",
443 description: "Hash listed fields (SHA-256 / BLAKE3) into stable, join-able tokens.",
444 schema_fn: || schema::<HashConfig>(),
445 compile_fn: |kind, config| {
446 let cfg = decode::<HashConfig>(kind, config)?;
447 let stage = TransformStage::Map(RecordTransform::Hash {
448 fields: cfg.fields,
449 algorithm: cfg.algorithm,
450 encoding: cfg.encoding,
451 salt: cfg.salt,
452 into: cfg.into,
453 });
454 validate_stage(kind, &stage)?;
455 Ok(stage)
456 },
457 },
458 TransformDef {
459 kind: "json_parse",
460 description: "Parse a stringified-JSON field into a real nested JSON value.",
461 schema_fn: || schema::<JsonParseConfig>(),
462 compile_fn: |kind, config| {
463 let cfg = decode::<JsonParseConfig>(kind, config)?;
464 let stage = TransformStage::Map(RecordTransform::JsonParse {
465 fields: cfg.fields,
466 on_error: cfg.on_error,
467 into: cfg.into,
468 });
469 validate_stage(kind, &stage)?;
470 Ok(stage)
471 },
472 },
473 TransformDef {
474 kind: "coalesce",
475 description: "Fill a missing/null field from a default or first non-null fallback key.",
476 schema_fn: || schema::<CoalesceConfig>(),
477 compile_fn: |kind, config| {
478 let cfg = decode::<CoalesceConfig>(kind, config)?;
479 let stage = TransformStage::Map(RecordTransform::Coalesce {
480 field: cfg.field,
481 default: cfg.default,
482 from: cfg.from,
483 treat_empty_string_as_null: cfg.treat_empty_string_as_null,
484 });
485 validate_stage(kind, &stage)?;
486 Ok(stage)
487 },
488 },
489 TransformDef {
490 kind: "split",
491 description: "Split a string field into an array on a delimiter.",
492 schema_fn: || schema::<SplitConfig>(),
493 compile_fn: |kind, config| {
494 let cfg = decode::<SplitConfig>(kind, config)?;
495 Ok(TransformStage::Map(RecordTransform::Split {
496 field: cfg.field,
497 delimiter: cfg.delimiter,
498 trim: cfg.trim,
499 into: cfg.into,
500 }))
501 },
502 },
503 TransformDef {
504 kind: "join",
505 description: "Join an array field into a string with a delimiter.",
506 schema_fn: || schema::<JoinConfig>(),
507 compile_fn: |kind, config| {
508 let cfg = decode::<JoinConfig>(kind, config)?;
509 Ok(TransformStage::Map(RecordTransform::Join {
510 field: cfg.field,
511 delimiter: cfg.delimiter,
512 into: cfg.into,
513 }))
514 },
515 },
516 #[cfg(feature = "transform-filter")]
517 TransformDef {
518 kind: "filter",
519 description: "Keep records where a JSONPath predicate is true.",
520 schema_fn: || schema::<FilterConfig>(),
521 compile_fn: |kind, config| {
522 let cfg = decode::<FilterConfig>(kind, config)?;
523 let stage = TransformStage::Filter(faucet_core::FilterSpec {
525 path: cfg.path,
526 op: cfg.op,
527 value: cfg.value,
528 });
529 faucet_core::compile_stage(&stage).map_err(|e| match e {
530 faucet_core::FaucetError::Transform(msg) => CliError::InvalidTransform {
531 name: kind.to_owned(),
532 message: msg,
533 },
534 other => CliError::InvalidTransform {
535 name: kind.to_owned(),
536 message: format!("{other}"),
537 },
538 })?;
539 Ok(stage)
540 },
541 },
542 #[cfg(feature = "transform-explode")]
543 TransformDef {
544 kind: "explode",
545 description: "Expand an array field into one record per element.",
546 schema_fn: || schema::<ExplodeConfig>(),
547 compile_fn: |kind, config| {
548 let cfg = decode::<ExplodeConfig>(kind, config)?;
549 let stage = TransformStage::Explode(faucet_core::ExplodeSpec {
550 path: cfg.path,
551 prefix: cfg.prefix,
552 separator: cfg.separator,
553 on_missing: cfg.on_missing,
554 });
555 faucet_core::compile_stage(&stage).map_err(|e| match e {
556 faucet_core::FaucetError::Transform(msg) => CliError::InvalidTransform {
557 name: kind.to_owned(),
558 message: msg,
559 },
560 other => CliError::InvalidTransform {
561 name: kind.to_owned(),
562 message: format!("{other}"),
563 },
564 })?;
565 Ok(stage)
566 },
567 },
568 ]);
569 }
570 #[cfg(feature = "transform-sql")]
571 {
572 defs.push(TransformDef {
573 kind: "sql",
574 description: "Run DuckDB SQL over the whole page; records are the `batch` relation.",
575 schema_fn: || schema_sql(),
576 compile_fn: |kind, config| {
577 let cfg: faucet_transform_sql::SqlTransformConfig = decode_sql(kind, config)?;
578 let transform = faucet_transform_sql::SqlTransform::compile(&cfg).map_err(|e| {
579 let message = match &e {
580 faucet_core::FaucetError::Transform(m)
581 | faucet_core::FaucetError::Config(m) => m.clone(),
582 other => format!("{other}"),
583 };
584 CliError::InvalidTransform {
585 name: kind.to_owned(),
586 message,
587 }
588 })?;
589 Ok(transform.into_page_stage())
590 },
591 });
592 }
593 #[cfg(feature = "transform-wasm")]
594 {
595 defs.push(TransformDef {
596 kind: "wasm",
597 description: "Run a user-provided sandboxed .wasm module over each record (wasmtime).",
598 schema_fn: || schema_wasm(),
599 compile_fn: |kind, config| {
600 let cfg: faucet_transform_wasm::WasmTransformConfig = decode_wasm(kind, config)?;
601 let transform =
602 faucet_transform_wasm::WasmTransform::compile(&cfg).map_err(|e| {
603 let message = match &e {
604 faucet_core::FaucetError::Transform(m)
605 | faucet_core::FaucetError::Config(m) => m.clone(),
606 other => format!("{other}"),
607 };
608 CliError::InvalidTransform {
609 name: kind.to_owned(),
610 message,
611 }
612 })?;
613 Ok(transform.into_page_stage())
614 },
615 });
616 }
617 #[cfg(feature = "transform-cdc-unwrap")]
618 {
619 defs.push(TransformDef {
620 kind: "cdc_unwrap",
621 description: "Normalize a CDC envelope into a flat row + delete marker (for upsert sinks).",
622 schema_fn: || schema::<CdcUnwrapConfig>(),
623 compile_fn: |kind, config| {
624 let cfg = decode::<CdcUnwrapConfig>(kind, config)?;
625 Ok(faucet_core::TransformStage::CdcUnwrap(faucet_core::CdcUnwrapSpec {
626 op_field: cfg.op_field,
627 after_field: cfg.after_field,
628 before_field: cfg.before_field,
629 key_field: cfg.key_field,
630 marker_field: cfg.marker_field,
631 delete_ops: cfg.delete_ops,
632 drop_ops: cfg.drop_ops,
633 }))
634 },
635 });
636 }
637 defs
638}
639
640pub fn compile_transforms(specs: &[TransformSpec]) -> CliResult<Vec<TransformStage>> {
645 let mut out = Vec::with_capacity(specs.len());
646 for s in specs {
647 out.push(compile_one(s)?);
648 }
649 Ok(out)
650}
651
652fn compile_one(spec: &TransformSpec) -> CliResult<TransformStage> {
653 match registry().into_iter().find(|t| t.kind == spec.kind) {
654 Some(def) => (def.compile_fn)(&spec.kind, spec.config.clone()),
655 None => Err(unknown_transform(&spec.kind)),
656 }
657}
658
659#[cfg(feature = "arrow")]
665pub fn compile_transforms_columnar(
666 specs: &[TransformSpec],
667) -> CliResult<(
668 Vec<TransformStage>,
669 Vec<Option<faucet_core::stage::PageFnBatchBox>>,
670)> {
671 let mut stages = Vec::with_capacity(specs.len());
672 let mut batches = Vec::with_capacity(specs.len());
673 for s in specs {
674 #[cfg(feature = "transform-sql")]
675 if s.kind == "sql" {
676 let cfg = decode_sql("sql", s.config.clone())?;
677 let transform = faucet_transform_sql::SqlTransform::compile(&cfg).map_err(|e| {
678 let message = match &e {
679 faucet_core::FaucetError::Transform(m)
680 | faucet_core::FaucetError::Config(m) => m.clone(),
681 other => format!("{other}"),
682 };
683 CliError::InvalidTransform {
684 name: "sql".to_owned(),
685 message,
686 }
687 })?;
688 let (stage, batch) = transform.into_columnar_stage();
689 stages.push(stage);
690 batches.push(Some(batch));
691 continue;
692 }
693 stages.push(compile_one(s)?);
694 batches.push(None);
695 }
696 Ok((stages, batches))
697}
698
699pub fn transform_descriptions() -> Vec<(&'static str, &'static str)> {
702 registry()
703 .into_iter()
704 .map(|t| (t.kind, t.description))
705 .collect()
706}
707
708pub fn available_transforms() -> Vec<&'static str> {
710 registry().into_iter().map(|t| t.kind).collect()
711}
712
713#[cfg(feature = "quality")]
718pub fn quality_descriptions() -> Vec<(&'static str, &'static str)> {
719 let mut checks = vec![
720 ("not_null", "field present and non-null"),
721 ("not_empty", "string non-empty after trim"),
722 ("regex_match", "string matches a regex"),
723 ("value_in_set", "value is in an allowed set"),
724 ("not_in_set", "value is not in a forbidden set"),
725 ("compare", "numeric/scalar comparison (gt/gte/lt/lte/eq/ne)"),
726 ("type_is", "value is of an expected JSON type"),
727 ("string_length", "string length within [min,max]"),
728 ];
729 #[cfg(feature = "quality-jsonschema")]
730 checks.push((
731 "json_schema",
732 "record validates against a JSON Schema (feature-gated)",
733 ));
734 checks.extend([
735 ("row_count", "batch row count within [min,max]"),
736 ("null_rate", "batch null rate of a field <= max"),
737 ("unique", "composite key unique within the batch"),
738 (
739 "distinct_count",
740 "distinct values of a field within [min,max]",
741 ),
742 ]);
743 checks
744}
745
746pub fn transform_schema(name: &str) -> CliResult<Value> {
750 registry()
751 .into_iter()
752 .find(|t| t.kind == name)
753 .map(|t| (t.schema_fn)())
754 .ok_or_else(|| unknown_transform(name))
755}
756
757fn unknown_transform(name: &str) -> CliError {
758 let available = available_transforms();
759 CliError::UnknownTransform {
760 name: name.to_owned(),
761 available: if available.is_empty() {
762 "(none — rebuild faucet-cli with the `transforms` feature enabled)".to_owned()
763 } else {
764 available.join(", ")
765 },
766 }
767}
768
769#[cfg(any(feature = "transforms", feature = "transform-cdc-unwrap"))]
770fn schema<T: JsonSchema>() -> Value {
771 serde_json::to_value(schema_for!(T)).unwrap_or_else(|_| serde_json::json!({"type": "object"}))
772}
773
774#[cfg(any(feature = "transforms", feature = "transform-cdc-unwrap"))]
775fn decode<T: serde::de::DeserializeOwned>(name: &str, config: Value) -> CliResult<T> {
776 serde_json::from_value(config).map_err(|e| CliError::InvalidTransform {
777 name: name.to_owned(),
778 message: e.to_string(),
779 })
780}
781
782#[cfg(feature = "transforms")]
787fn validate_stage(kind: &str, stage: &TransformStage) -> CliResult<()> {
788 faucet_core::compile_stage(stage).map(|_| ()).map_err(|e| {
789 let message = match e {
790 faucet_core::FaucetError::Transform(m) | faucet_core::FaucetError::Config(m) => m,
791 other => format!("{other}"),
792 };
793 CliError::InvalidTransform {
794 name: kind.to_owned(),
795 message,
796 }
797 })
798}
799
800#[cfg(feature = "transform-sql")]
801fn schema_sql() -> Value {
802 serde_json::to_value(faucet_core::schema_for!(
803 faucet_transform_sql::SqlTransformConfig
804 ))
805 .unwrap_or(Value::Null)
806}
807
808#[cfg(feature = "transform-sql")]
809fn decode_sql(kind: &str, config: Value) -> CliResult<faucet_transform_sql::SqlTransformConfig> {
810 serde_json::from_value(config).map_err(|e| CliError::InvalidTransform {
811 name: kind.to_owned(),
812 message: e.to_string(),
813 })
814}
815
816#[cfg(feature = "transform-wasm")]
817fn schema_wasm() -> Value {
818 serde_json::to_value(faucet_core::schema_for!(
819 faucet_transform_wasm::WasmTransformConfig
820 ))
821 .unwrap_or(Value::Null)
822}
823
824#[cfg(feature = "transform-wasm")]
825fn decode_wasm(kind: &str, config: Value) -> CliResult<faucet_transform_wasm::WasmTransformConfig> {
826 serde_json::from_value(config).map_err(|e| CliError::InvalidTransform {
827 name: kind.to_owned(),
828 message: e.to_string(),
829 })
830}
831
832#[cfg(test)]
833mod tests {
834 use super::*;
835 use serde_json::json;
836
837 #[test]
838 fn empty_list_compiles_to_empty() {
839 let out = compile_transforms(&[]).unwrap();
840 assert!(out.is_empty());
841 }
842
843 #[cfg(feature = "transforms")]
844 #[test]
845 fn compiles_keys_case_and_flatten() {
846 let specs = vec![
847 TransformSpec {
848 kind: "keys_case".into(),
849 config: json!({"mode": "snake"}),
850 },
851 TransformSpec {
852 kind: "flatten".into(),
853 config: json!({"separator": "."}),
854 },
855 ];
856 let out = compile_transforms(&specs).unwrap();
857 assert_eq!(out.len(), 2);
858 }
859
860 #[cfg(feature = "transforms")]
861 #[test]
862 fn keys_case_rejects_unknown_mode() {
863 let specs = vec![TransformSpec {
864 kind: "keys_case".into(),
865 config: json!({"mode": "spongebob"}),
866 }];
867 let err = compile_transforms(&specs).unwrap_err();
868 match err {
869 CliError::InvalidTransform { name, .. } => assert_eq!(name, "keys_case"),
870 other => panic!("expected InvalidTransform, got {other:?}"),
871 }
872 }
873
874 #[cfg(feature = "transforms")]
875 #[test]
876 fn keys_case_requires_mode() {
877 let specs = vec![TransformSpec {
878 kind: "keys_case".into(),
879 config: json!({}),
880 }];
881 let err = compile_transforms(&specs).unwrap_err();
882 match err {
883 CliError::InvalidTransform { name, .. } => assert_eq!(name, "keys_case"),
884 other => panic!("expected InvalidTransform, got {other:?}"),
885 }
886 }
887
888 #[cfg(feature = "transforms")]
889 #[test]
890 fn snake_case_kind_is_no_longer_recognized() {
891 let specs = vec![TransformSpec {
893 kind: "snake_case".into(),
894 config: json!({}),
895 }];
896 let err = compile_transforms(&specs).unwrap_err();
897 match err {
898 CliError::UnknownTransform { name, .. } => assert_eq!(name, "snake_case"),
899 other => panic!("expected UnknownTransform, got {other:?}"),
900 }
901 }
902
903 #[cfg(feature = "transforms")]
904 #[test]
905 fn rename_keys_requires_pattern_and_replacement() {
906 let specs = vec![TransformSpec {
907 kind: "rename_keys".into(),
908 config: json!({"pattern": "^_"}),
909 }];
910 let err = compile_transforms(&specs).unwrap_err();
911 match err {
912 CliError::InvalidTransform { name, .. } => assert_eq!(name, "rename_keys"),
913 other => panic!("expected InvalidTransform, got {other:?}"),
914 }
915 }
916
917 #[test]
918 fn unknown_transform_errors() {
919 let specs = vec![TransformSpec {
920 kind: "make_uppercase".into(),
921 config: json!({}),
922 }];
923 let err = compile_transforms(&specs).unwrap_err();
924 match err {
925 CliError::UnknownTransform { name, .. } => assert_eq!(name, "make_uppercase"),
926 other => panic!("expected UnknownTransform, got {other:?}"),
927 }
928 }
929
930 #[cfg(feature = "transforms")]
931 #[test]
932 fn compiles_select_and_drop() {
933 let specs = vec![
934 TransformSpec {
935 kind: "select".into(),
936 config: json!({"fields": ["id", "name"]}),
937 },
938 TransformSpec {
939 kind: "drop".into(),
940 config: json!({"fields": ["secret"]}),
941 },
942 ];
943 let out = compile_transforms(&specs).unwrap();
944 assert_eq!(out.len(), 2);
945 }
946
947 #[cfg(feature = "transforms")]
948 #[test]
949 fn compiles_set_with_object_values() {
950 let specs = vec![TransformSpec {
951 kind: "set".into(),
952 config: json!({"values": {"_source": "api", "version": 1}}),
953 }];
954 let out = compile_transforms(&specs).unwrap();
955 assert_eq!(out.len(), 1);
956 }
957
958 #[cfg(feature = "transforms")]
959 #[test]
960 fn compiles_rename_field() {
961 let specs = vec![TransformSpec {
962 kind: "rename_field".into(),
963 config: json!({"fields": {"old": "new"}}),
964 }];
965 let out = compile_transforms(&specs).unwrap();
966 assert_eq!(out.len(), 1);
967 }
968
969 #[cfg(feature = "transforms")]
970 #[test]
971 fn compiles_cast_with_default_on_error() {
972 let specs = vec![TransformSpec {
973 kind: "cast".into(),
974 config: json!({"fields": {"age": "int", "price": "float"}}),
975 }];
976 let out = compile_transforms(&specs).unwrap();
977 assert_eq!(out.len(), 1);
978 }
979
980 #[cfg(feature = "transforms")]
981 #[test]
982 fn cast_rejects_unknown_target_type() {
983 let specs = vec![TransformSpec {
984 kind: "cast".into(),
985 config: json!({"fields": {"x": "uuid"}}),
986 }];
987 let err = compile_transforms(&specs).unwrap_err();
988 match err {
989 CliError::InvalidTransform { name, .. } => assert_eq!(name, "cast"),
990 other => panic!("expected InvalidTransform, got {other:?}"),
991 }
992 }
993
994 #[cfg(feature = "transforms")]
995 #[test]
996 fn cast_rejects_unknown_on_error_mode() {
997 let specs = vec![TransformSpec {
998 kind: "cast".into(),
999 config: json!({"fields": {"x": "int"}, "on_error": "explode"}),
1000 }];
1001 let err = compile_transforms(&specs).unwrap_err();
1002 match err {
1003 CliError::InvalidTransform { name, .. } => assert_eq!(name, "cast"),
1004 other => panic!("expected InvalidTransform, got {other:?}"),
1005 }
1006 }
1007
1008 #[cfg(feature = "transforms")]
1009 #[test]
1010 fn redact_uses_default_mask_when_omitted() {
1011 let specs = vec![TransformSpec {
1012 kind: "redact".into(),
1013 config: json!({"fields": ["ssn"]}),
1014 }];
1015 let out = compile_transforms(&specs).unwrap();
1016 assert_eq!(out.len(), 1);
1017 }
1018
1019 #[cfg(feature = "transforms")]
1020 #[test]
1021 fn compiles_hash_with_defaults_and_options() {
1022 let specs = vec![
1023 TransformSpec {
1024 kind: "hash".into(),
1025 config: json!({"fields": ["email"]}),
1026 },
1027 TransformSpec {
1028 kind: "hash".into(),
1029 config: json!({
1030 "fields": ["user_id"],
1031 "algorithm": "blake3",
1032 "encoding": "base64",
1033 "salt": "pepper",
1034 "into": "user_id_hash"
1035 }),
1036 },
1037 ];
1038 let out = compile_transforms(&specs).unwrap();
1039 assert_eq!(out.len(), 2);
1040 }
1041
1042 #[cfg(feature = "transforms")]
1043 #[test]
1044 fn hash_empty_fields_is_rejected_at_load() {
1045 let specs = vec![TransformSpec {
1046 kind: "hash".into(),
1047 config: json!({"fields": []}),
1048 }];
1049 match compile_transforms(&specs).unwrap_err() {
1050 CliError::InvalidTransform { name, .. } => assert_eq!(name, "hash"),
1051 other => panic!("expected InvalidTransform, got {other:?}"),
1052 }
1053 }
1054
1055 #[cfg(feature = "transforms")]
1056 #[test]
1057 fn hash_into_with_multiple_fields_is_rejected_at_load() {
1058 let specs = vec![TransformSpec {
1059 kind: "hash".into(),
1060 config: json!({"fields": ["a", "b"], "into": "x"}),
1061 }];
1062 match compile_transforms(&specs).unwrap_err() {
1063 CliError::InvalidTransform { name, message } => {
1064 assert_eq!(name, "hash");
1065 assert!(message.contains("into"), "{message}");
1066 }
1067 other => panic!("expected InvalidTransform, got {other:?}"),
1068 }
1069 }
1070
1071 #[cfg(feature = "transforms")]
1072 #[test]
1073 fn compiles_json_parse_with_on_error() {
1074 let specs = vec![TransformSpec {
1075 kind: "json_parse".into(),
1076 config: json!({"fields": ["payload"], "on_error": "null", "into": "parsed"}),
1077 }];
1078 let out = compile_transforms(&specs).unwrap();
1079 assert_eq!(out.len(), 1);
1080 }
1081
1082 #[cfg(feature = "transforms")]
1083 #[test]
1084 fn json_parse_into_with_multiple_fields_is_rejected_at_load() {
1085 let specs = vec![TransformSpec {
1086 kind: "json_parse".into(),
1087 config: json!({"fields": ["a", "b"], "into": "x"}),
1088 }];
1089 match compile_transforms(&specs).unwrap_err() {
1090 CliError::InvalidTransform { name, .. } => assert_eq!(name, "json_parse"),
1091 other => panic!("expected InvalidTransform, got {other:?}"),
1092 }
1093 }
1094
1095 #[cfg(feature = "transforms")]
1096 #[test]
1097 fn compiles_coalesce_with_default() {
1098 let specs = vec![TransformSpec {
1099 kind: "coalesce".into(),
1100 config: json!({"field": "status", "default": "unknown"}),
1101 }];
1102 let out = compile_transforms(&specs).unwrap();
1103 assert_eq!(out.len(), 1);
1104 }
1105
1106 #[cfg(feature = "transforms")]
1107 #[test]
1108 fn compiles_coalesce_with_from() {
1109 let specs = vec![TransformSpec {
1110 kind: "coalesce".into(),
1111 config: json!({"field": "status", "from": ["status", "state"]}),
1112 }];
1113 let out = compile_transforms(&specs).unwrap();
1114 assert_eq!(out.len(), 1);
1115 }
1116
1117 #[cfg(feature = "transforms")]
1118 #[test]
1119 fn coalesce_both_default_and_from_is_rejected_at_load() {
1120 let specs = vec![TransformSpec {
1121 kind: "coalesce".into(),
1122 config: json!({"field": "s", "default": "x", "from": ["y"]}),
1123 }];
1124 match compile_transforms(&specs).unwrap_err() {
1125 CliError::InvalidTransform { name, .. } => assert_eq!(name, "coalesce"),
1126 other => panic!("expected InvalidTransform, got {other:?}"),
1127 }
1128 }
1129
1130 #[cfg(feature = "transforms")]
1131 #[test]
1132 fn coalesce_neither_default_nor_from_is_rejected_at_load() {
1133 let specs = vec![TransformSpec {
1134 kind: "coalesce".into(),
1135 config: json!({"field": "s"}),
1136 }];
1137 match compile_transforms(&specs).unwrap_err() {
1138 CliError::InvalidTransform { name, .. } => assert_eq!(name, "coalesce"),
1139 other => panic!("expected InvalidTransform, got {other:?}"),
1140 }
1141 }
1142
1143 #[cfg(feature = "transforms")]
1144 #[test]
1145 fn compiles_split_and_join() {
1146 let specs = vec![
1147 TransformSpec {
1148 kind: "split".into(),
1149 config: json!({"field": "tags", "delimiter": ",", "trim": true}),
1150 },
1151 TransformSpec {
1152 kind: "join".into(),
1153 config: json!({"field": "tags", "delimiter": ",", "into": "csv"}),
1154 },
1155 ];
1156 let out = compile_transforms(&specs).unwrap();
1157 assert_eq!(out.len(), 2);
1158 }
1159
1160 #[cfg(feature = "transforms")]
1161 #[test]
1162 fn value_case_requires_mode() {
1163 let specs = vec![TransformSpec {
1164 kind: "value_case".into(),
1165 config: json!({"fields": ["email"]}),
1166 }];
1167 let err = compile_transforms(&specs).unwrap_err();
1168 match err {
1169 CliError::InvalidTransform { name, .. } => assert_eq!(name, "value_case"),
1170 other => panic!("expected InvalidTransform, got {other:?}"),
1171 }
1172 }
1173
1174 #[cfg(feature = "transforms")]
1175 #[test]
1176 fn available_transforms_lists_every_kind() {
1177 let names = available_transforms();
1178 for expected in [
1179 "flatten",
1180 "rename_keys",
1181 "keys_case",
1182 "select",
1183 "drop",
1184 "set",
1185 "rename_field",
1186 "cast",
1187 "redact",
1188 "value_case",
1189 "spell_symbols",
1190 "hash",
1191 "json_parse",
1192 "coalesce",
1193 "split",
1194 "join",
1195 "filter",
1196 "explode",
1197 ] {
1198 assert!(names.contains(&expected), "missing {expected}");
1199 }
1200 assert!(
1201 !names.contains(&"snake_case"),
1202 "snake_case must be removed in favour of keys_case"
1203 );
1204 }
1205
1206 #[cfg(feature = "transforms")]
1207 #[test]
1208 fn transform_descriptions_covers_every_compiled_kind() {
1209 let names = available_transforms();
1212 let desc_names: Vec<&'static str> = transform_descriptions()
1213 .into_iter()
1214 .map(|(n, _)| n)
1215 .collect();
1216 assert_eq!(names, desc_names);
1217 for (_, desc) in transform_descriptions() {
1218 assert!(!desc.is_empty(), "every transform needs a description");
1219 }
1220 }
1221
1222 #[cfg(feature = "transforms")]
1223 #[test]
1224 fn transform_schema_returns_object_for_every_kind() {
1225 for name in available_transforms() {
1226 let schema = transform_schema(name).unwrap_or_else(|e| {
1227 panic!("schema lookup failed for {name}: {e}");
1228 });
1229 assert!(schema.is_object(), "schema for {name} must be an object");
1230 }
1231 }
1232
1233 #[cfg(feature = "transforms")]
1234 #[test]
1235 fn transform_schema_select_and_drop_share_shape() {
1236 let select = transform_schema("select").unwrap();
1239 let drop = transform_schema("drop").unwrap();
1240 assert_eq!(select, drop);
1241 }
1242
1243 #[test]
1244 fn transform_schema_unknown_errors_with_available_list() {
1245 let err = transform_schema("make_uppercase").unwrap_err();
1246 match err {
1247 CliError::UnknownTransform { name, available } => {
1248 assert_eq!(name, "make_uppercase");
1249 #[cfg(feature = "transforms")]
1250 assert!(available.contains("flatten"), "{available}");
1251 #[cfg(not(feature = "transforms"))]
1252 assert!(available.contains("rebuild"), "{available}");
1253 }
1254 other => panic!("expected UnknownTransform, got {other:?}"),
1255 }
1256 }
1257
1258 #[cfg(feature = "transform-filter")]
1259 #[test]
1260 fn compiles_filter_eq() {
1261 let specs = vec![TransformSpec {
1262 kind: "filter".into(),
1263 config: json!({"path": "status", "op": "eq", "value": "active"}),
1264 }];
1265 let out = compile_transforms(&specs).unwrap();
1266 assert_eq!(out.len(), 1);
1267 assert!(matches!(out[0], TransformStage::Filter(_)));
1268 }
1269
1270 #[cfg(feature = "transform-filter")]
1271 #[test]
1272 fn filter_rejects_in_with_non_array_value() {
1273 let specs = vec![TransformSpec {
1274 kind: "filter".into(),
1275 config: json!({"path": "v", "op": "in", "value": "scalar"}),
1276 }];
1277 let err = compile_transforms(&specs).unwrap_err();
1278 match err {
1279 CliError::InvalidTransform { name, message } => {
1280 assert_eq!(name, "filter");
1281 assert!(message.contains("requires an array"), "{message}");
1282 }
1283 other => panic!("expected InvalidTransform, got {other:?}"),
1284 }
1285 }
1286
1287 #[cfg(feature = "transform-filter")]
1288 #[test]
1289 fn filter_rejects_exists_with_value() {
1290 let specs = vec![TransformSpec {
1291 kind: "filter".into(),
1292 config: json!({"path": "v", "op": "exists", "value": "x"}),
1293 }];
1294 let err = compile_transforms(&specs).unwrap_err();
1295 match err {
1296 CliError::InvalidTransform { name, .. } => assert_eq!(name, "filter"),
1297 other => panic!("expected InvalidTransform, got {other:?}"),
1298 }
1299 }
1300
1301 #[cfg(feature = "transform-filter")]
1302 #[test]
1303 fn filter_rejects_bad_path() {
1304 let specs = vec![TransformSpec {
1305 kind: "filter".into(),
1306 config: json!({"path": "$..items", "op": "exists"}),
1307 }];
1308 let err = compile_transforms(&specs).unwrap_err();
1309 match err {
1310 CliError::InvalidTransform { name, .. } => assert_eq!(name, "filter"),
1311 other => panic!("expected InvalidTransform, got {other:?}"),
1312 }
1313 }
1314
1315 #[cfg(feature = "transform-explode")]
1316 #[test]
1317 fn compiles_explode_with_defaults() {
1318 let specs = vec![TransformSpec {
1319 kind: "explode".into(),
1320 config: json!({"path": "items"}),
1321 }];
1322 let out = compile_transforms(&specs).unwrap();
1323 assert_eq!(out.len(), 1);
1324 assert!(matches!(out[0], TransformStage::Explode(_)));
1325 }
1326
1327 #[cfg(feature = "transform-explode")]
1328 #[test]
1329 fn compiles_explode_with_custom_prefix_and_on_missing() {
1330 let specs = vec![TransformSpec {
1331 kind: "explode".into(),
1332 config: json!({
1333 "path": "items",
1334 "prefix": "item",
1335 "separator": "_",
1336 "on_missing": "drop"
1337 }),
1338 }];
1339 let out = compile_transforms(&specs).unwrap();
1340 assert_eq!(out.len(), 1);
1341 }
1342
1343 #[cfg(feature = "transform-explode")]
1344 #[test]
1345 fn explode_rejects_bad_path() {
1346 let specs = vec![TransformSpec {
1347 kind: "explode".into(),
1348 config: json!({"path": "$..items"}),
1349 }];
1350 let err = compile_transforms(&specs).unwrap_err();
1351 match err {
1352 CliError::InvalidTransform { name, .. } => assert_eq!(name, "explode"),
1353 other => panic!("expected InvalidTransform, got {other:?}"),
1354 }
1355 }
1356
1357 #[cfg(feature = "transform-explode")]
1358 #[test]
1359 fn explode_rejects_invalid_on_missing() {
1360 let specs = vec![TransformSpec {
1361 kind: "explode".into(),
1362 config: json!({"path": "items", "on_missing": "explode_harder"}),
1363 }];
1364 let err = compile_transforms(&specs).unwrap_err();
1365 match err {
1366 CliError::InvalidTransform { name, .. } => assert_eq!(name, "explode"),
1367 other => panic!("expected InvalidTransform, got {other:?}"),
1368 }
1369 }
1370
1371 #[cfg(feature = "transform-sql")]
1372 #[test]
1373 fn compiles_sql_to_page_fn() {
1374 let specs = vec![TransformSpec {
1375 kind: "sql".into(),
1376 config: json!({"query": "SELECT * FROM batch"}),
1377 }];
1378 let out = compile_transforms(&specs).unwrap();
1379 assert_eq!(out.len(), 1);
1380 assert!(matches!(out[0], faucet_core::TransformStage::PageFn(_)));
1381 }
1382
1383 #[cfg(feature = "transform-sql")]
1384 #[test]
1385 fn sql_bad_query_is_invalid_transform() {
1386 let specs = vec![TransformSpec {
1387 kind: "sql".into(),
1388 config: json!({"query": "SELEKT bad"}),
1389 }];
1390 let err = compile_transforms(&specs).unwrap_err();
1391 match err {
1392 CliError::InvalidTransform { name, .. } => assert_eq!(name, "sql"),
1393 other => panic!("expected InvalidTransform, got {other:?}"),
1394 }
1395 }
1396
1397 #[cfg(feature = "transform-sql")]
1398 #[test]
1399 fn sql_schema_and_listing_present() {
1400 assert!(transform_schema("sql").is_ok());
1401 assert!(available_transforms().contains(&"sql"));
1402 }
1403
1404 #[cfg(feature = "transform-wasm")]
1405 #[test]
1406 fn wasm_schema_and_listing_present() {
1407 assert!(transform_schema("wasm").is_ok());
1408 assert!(available_transforms().contains(&"wasm"));
1409 }
1410
1411 #[cfg(feature = "transform-wasm")]
1412 #[test]
1413 fn wasm_missing_module_field_is_invalid_transform() {
1414 let specs = vec![TransformSpec {
1415 kind: "wasm".into(),
1416 config: json!({}),
1417 }];
1418 let err = compile_transforms(&specs).unwrap_err();
1419 match err {
1420 CliError::InvalidTransform { name, .. } => assert_eq!(name, "wasm"),
1421 other => panic!("expected InvalidTransform, got {other:?}"),
1422 }
1423 }
1424
1425 #[cfg(feature = "transform-wasm")]
1426 #[test]
1427 fn wasm_nonexistent_module_is_invalid_transform() {
1428 let specs = vec![TransformSpec {
1429 kind: "wasm".into(),
1430 config: json!({"module": "/no/such/path/mod.wasm"}),
1431 }];
1432 let err = compile_transforms(&specs).unwrap_err();
1433 match err {
1434 CliError::InvalidTransform { name, message } => {
1435 assert_eq!(name, "wasm");
1436 assert!(message.contains("cannot read module"), "{message}");
1437 }
1438 other => panic!("expected InvalidTransform, got {other:?}"),
1439 }
1440 }
1441
1442 #[cfg(feature = "transform-cdc-unwrap")]
1443 #[test]
1444 fn compiles_cdc_unwrap_with_defaults() {
1445 let specs = vec![TransformSpec {
1446 kind: "cdc_unwrap".into(),
1447 config: json!({}),
1448 }];
1449 let out = compile_transforms(&specs).unwrap();
1450 assert_eq!(out.len(), 1);
1451 assert!(matches!(out[0], faucet_core::TransformStage::CdcUnwrap(_)));
1452 }
1453
1454 #[cfg(feature = "transform-cdc-unwrap")]
1455 #[test]
1456 fn cdc_unwrap_schema_and_listing_present() {
1457 assert!(transform_schema("cdc_unwrap").is_ok());
1458 assert!(available_transforms().contains(&"cdc_unwrap"));
1459 }
1460
1461 #[cfg(feature = "quality")]
1462 #[test]
1463 fn quality_descriptions_has_one_entry_per_check() {
1464 #[cfg(feature = "quality-jsonschema")]
1470 assert_eq!(quality_descriptions().len(), 13);
1471 #[cfg(not(feature = "quality-jsonschema"))]
1472 assert_eq!(quality_descriptions().len(), 12);
1473 }
1474}