diff --git a/crates/adapterlib/src/catalog.rs b/crates/adapterlib/src/catalog.rs index 6260edc6c33..a3fb5d031ef 100644 --- a/crates/adapterlib/src/catalog.rs +++ b/crates/adapterlib/src/catalog.rs @@ -31,6 +31,7 @@ use crate::errors::controller::ControllerError; use crate::format::InputBuffer; use crate::postprocess::PostprocessorRegistry; use crate::preprocess::PreprocessorRegistry; +use crate::transport::{InputTransportRegistry, OutputTransportRegistry}; /// Descriptor that specifies the format in which records are received /// or into which they should be encoded before sending. @@ -1166,6 +1167,12 @@ pub trait CircuitCatalog: Send + Sync { /// The registry used to insert new user-defined postprocessors fn postprocessor_registry(&self) -> Arc>; + + /// The registry used to look up input transport endpoint factories. + fn input_transport_registry(&self) -> Arc>; + + /// The registry used to look up output transport endpoint factories. + fn output_transport_registry(&self) -> Arc>; } #[doc(hidden)] diff --git a/crates/adapterlib/src/transport.rs b/crates/adapterlib/src/transport.rs index a5b3c81cc1b..5574fc35ac7 100644 --- a/crates/adapterlib/src/transport.rs +++ b/crates/adapterlib/src/transport.rs @@ -2,14 +2,14 @@ use anyhow::{Error as AnyError, Result as AnyResult}; use chrono::{DateTime, Utc}; use dyn_clone::DynClone; use feldera_types::adapter_stats::ConnectorHealth; -use feldera_types::config::FtModel; +use feldera_types::config::{FtModel, TransportConfig}; use feldera_types::coordination::Completion; use feldera_types::program_schema::Relation; use rmpv::{Value as RmpValue, ext::Error as RmpDecodeError}; use serde::Deserialize; use serde::de::DeserializeOwned; use serde_json::Value as JsonValue; -use std::collections::VecDeque; +use std::collections::{BTreeMap, VecDeque}; use std::fmt::Display; use std::marker::PhantomData; use std::sync::atomic::{AtomicBool, Ordering}; @@ -75,6 +75,50 @@ pub trait TransportInputEndpoint: InputEndpoint { ) -> AnyResult>; } +/// Factory for creating input transport endpoints from transport configuration. +pub trait InputTransportEndpointFactory: Send + Sync { + fn create( + &self, + config: &TransportConfig, + ) -> AnyResult>>; +} + +/// Registry of input transport endpoint factories keyed by transport name. +#[derive(Default)] +pub struct InputTransportRegistry { + registered: BTreeMap>, +} + +impl InputTransportRegistry { + pub fn new() -> Self { + Self { + registered: BTreeMap::new(), + } + } + + pub fn register( + &mut self, + name: impl Into, + factory: Box, + ) { + self.registered.insert(name.into(), Arc::from(factory)); + } + + pub fn get(&self, name: &str) -> Option> { + self.registered.get(name).cloned() + } + + pub fn create_endpoint( + &self, + config: &TransportConfig, + ) -> AnyResult>> { + let Some(factory) = self.get(config.name()) else { + return Ok(None); + }; + factory.create(config) + } +} + #[doc(hidden)] pub trait IntegratedInputEndpoint: InputEndpoint { fn open( @@ -1094,6 +1138,54 @@ pub trait OutputEndpoint: Send { } } +/// Factory for creating output transport endpoints from transport configuration. +pub trait OutputTransportEndpointFactory: Send + Sync { + fn create( + &self, + config: &TransportConfig, + endpoint_name: &str, + fault_tolerant: bool, + ) -> AnyResult>>; +} + +/// Registry of output transport endpoint factories keyed by transport name. +#[derive(Default)] +pub struct OutputTransportRegistry { + registered: BTreeMap>, +} + +impl OutputTransportRegistry { + pub fn new() -> Self { + Self { + registered: BTreeMap::new(), + } + } + + pub fn register( + &mut self, + name: impl Into, + factory: Box, + ) { + self.registered.insert(name.into(), Arc::from(factory)); + } + + pub fn get(&self, name: &str) -> Option> { + self.registered.get(name).cloned() + } + + pub fn create_endpoint( + &self, + config: &TransportConfig, + endpoint_name: &str, + fault_tolerant: bool, + ) -> AnyResult>> { + let Some(factory) = self.get(config.name()) else { + return Ok(None); + }; + factory.create(config, endpoint_name, fault_tolerant) + } +} + /// An [UnboundedReceiver] wrapper for [InputReaderCommand] for fault-tolerant connectors. /// /// A fault-tolerant connector wants to receive, in order: diff --git a/crates/adapters/src/adhoc/table.rs b/crates/adapters/src/adhoc/table.rs index e0a0ae54bb9..d52b6e56c65 100644 --- a/crates/adapters/src/adhoc/table.rs +++ b/crates/adapters/src/adhoc/table.rs @@ -293,7 +293,7 @@ impl DataSink for AdHocTableSink { let config = InputEndpointConfig::new( self.name.to_string(), ConnectorConfig::new( - TransportConfig::AdHocInput(config), + TransportConfig::new(TransportConfig::ADHOC_INPUT, config), Some(FormatConfig { name: Cow::from("parquet"), config: serde_json::Value::Null, diff --git a/crates/adapters/src/catalog.rs b/crates/adapters/src/catalog.rs index 2b367b14a1a..5fdb70092b8 100644 --- a/crates/adapters/src/catalog.rs +++ b/crates/adapters/src/catalog.rs @@ -1,6 +1,8 @@ use feldera_adapterlib::{ - errors::controller::ControllerError, postprocess::PostprocessorRegistry, + errors::controller::ControllerError, + postprocess::PostprocessorRegistry, preprocess::PreprocessorRegistry, + transport::{InputTransportRegistry, OutputTransportRegistry}, }; use feldera_types::program_schema::SqlIdentifier; use std::{ @@ -14,6 +16,8 @@ pub use feldera_adapterlib::catalog::*; pub struct Catalog { input_collection_handles: BTreeMap, output_batch_handles: BTreeMap, + input_transport_registry: Arc>, + output_transport_registry: Arc>, preprocessor_registry: Arc>, postprocessor_registry: Arc>, } @@ -29,6 +33,12 @@ impl Catalog { Self { input_collection_handles: BTreeMap::new(), output_batch_handles: BTreeMap::new(), + input_transport_registry: Arc::new(Mutex::new( + crate::transport::builtin_input_transport_registry(), + )), + output_transport_registry: Arc::new(Mutex::new( + crate::transport::builtin_output_transport_registry(), + )), preprocessor_registry: Arc::new(Mutex::new(PreprocessorRegistry::new())), postprocessor_registry: Arc::new(Mutex::new(PostprocessorRegistry::new())), } @@ -128,4 +138,12 @@ impl CircuitCatalog for Catalog { fn postprocessor_registry(&self) -> Arc> { self.postprocessor_registry.clone() } + + fn input_transport_registry(&self) -> Arc> { + self.input_transport_registry.clone() + } + + fn output_transport_registry(&self) -> Arc> { + self.output_transport_registry.clone() + } } diff --git a/crates/adapters/src/controller.rs b/crates/adapters/src/controller.rs index e90ee69795d..d5e53da4470 100644 --- a/crates/adapters/src/controller.rs +++ b/crates/adapters/src/controller.rs @@ -35,7 +35,6 @@ use crate::server::metrics::{HistogramDiv, LabelStack, MetricsFormatter, Metrics use crate::server::{InitializationState, ServerState}; use crate::transport::Step; use crate::transport::clock::now_endpoint_config; -use crate::transport::{input_transport_config_to_endpoint, output_transport_config_to_endpoint}; use crate::util::{LongOperationWarning, run_on_thread_pool}; use crate::{ CircuitCatalog, Encoder, InputConsumer, OutputConsumer, OutputEndpoint, ParseError, @@ -94,7 +93,9 @@ use feldera_types::coordination::{ use feldera_types::format::json::JsonLines; use feldera_types::pipeline_diff::PipelineDiff; use feldera_types::runtime_status::BootstrapPolicy; -use feldera_types::secret_resolver::resolve_secret_references_in_connector_config; +use feldera_types::secret_resolver::{ + resolve_secret_references_in_connector_config, resolve_secret_references_via_json, +}; use feldera_types::suspend::{PermanentSuspendError, SuspendError, TemporarySuspendError}; use feldera_types::time_series::SampleStatistics; use feldera_types::transaction::{StartTransactionResponse, TransactionId}; @@ -4841,10 +4842,7 @@ impl ControllerInit { .filter(|(_, config)| { // The clock input connector will be automatically recreated and initialized // with the clock resolution from the pipeline config. - !matches!( - config.connector_config.transport, - TransportConfig::ClockInput(_) - ) + config.connector_config.transport.name() != TransportConfig::CLOCK }) .collect() } else { @@ -6268,12 +6266,23 @@ impl ControllerInner { endpoint_config: &InputEndpointConfig, resume_info: Option<(JsonValue, CheckpointInputEndpointMetrics)>, ) -> Result { - let endpoint = input_transport_config_to_endpoint( - &endpoint_config.connector_config.transport, - endpoint_name, + let transport_config = resolve_secret_references_via_json( &self.secrets_dir, + &endpoint_config.connector_config.transport, ) .map_err(|e| ControllerError::input_transport_error(endpoint_name, true, e))?; + let factory = self + .catalog + .input_transport_registry() + .lock() + .unwrap() + .get(transport_config.name()); + let endpoint = match factory { + Some(factory) => factory + .create(&transport_config) + .map_err(|e| ControllerError::input_transport_error(endpoint_name, true, e))?, + None => None, + }; // If `endpoint` is `None`, it means that the endpoint config specifies an integrated // input connector. Such endpoints are instantiated inside `add_input_endpoint`. @@ -6400,17 +6409,19 @@ impl ControllerInner { &resolved_connector_config.transport, &resolved_connector_config.format, ) { - (TransportConfig::Datagen(_), None) => FormatConfig { - name: Cow::from("json"), - config: serde_json::to_value(JsonParserConfig { - update_format: JsonUpdateFormat::Raw, - json_flavor: JsonFlavor::Datagen, - array: true, - lines: JsonLines::Multiple, - }) - .unwrap(), - }, - (TransportConfig::Datagen(_), Some(_)) => { + (transport, None) if transport.name() == TransportConfig::DATAGEN => { + FormatConfig { + name: Cow::from("json"), + config: serde_json::to_value(JsonParserConfig { + update_format: JsonUpdateFormat::Raw, + json_flavor: JsonFlavor::Datagen, + array: true, + lines: JsonLines::Multiple, + }) + .unwrap(), + } + } + (transport, Some(_)) if transport.name() == TransportConfig::DATAGEN => { return Err(ControllerError::input_format_not_supported( endpoint_name, "datagen endpoints do not support custom formats: remove the 'format' section from connector specification", @@ -6588,13 +6599,27 @@ impl ControllerInner { endpoint_config: &OutputEndpointConfig, initial_statistics: Option<&CheckpointOutputEndpointMetrics>, ) -> Result { - let endpoint = output_transport_config_to_endpoint( - &endpoint_config.connector_config.transport, - endpoint_name, - self.fault_tolerance == Some(FtModel::ExactlyOnce), + let transport_config = resolve_secret_references_via_json( &self.secrets_dir, + &endpoint_config.connector_config.transport, ) .map_err(|e| ControllerError::output_transport_error(endpoint_name, true, e))?; + let factory = self + .catalog + .output_transport_registry() + .lock() + .unwrap() + .get(transport_config.name()); + let endpoint = match factory { + Some(factory) => factory + .create( + &transport_config, + endpoint_name, + self.fault_tolerance == Some(FtModel::ExactlyOnce), + ) + .map_err(|e| ControllerError::output_transport_error(endpoint_name, true, e))?, + None => None, + }; // If `endpoint` is `None`, it means that the endpoint config specifies an integrated // output connector. Such endpoints are instantiated inside `add_output_endpoint`. diff --git a/crates/adapters/src/format/avro/output.rs b/crates/adapters/src/format/avro/output.rs index a15ce3e0075..cf414c90968 100644 --- a/crates/adapters/src/format/avro/output.rs +++ b/crates/adapters/src/format/avro/output.rs @@ -92,19 +92,25 @@ impl OutputFormat for AvroOutputFormat { ) })?; - if matches!( - config.transport, - feldera_types::config::TransportConfig::RedisOutput(_) - ) { + if config.transport.name() == TransportConfig::REDIS_OUTPUT { return Err(ControllerError::invalid_encoder_configuration( endpoint_name, "'avro' format not yet supported with Redis connector", )); } - let topic = match &config.transport { - TransportConfig::KafkaOutput(kafka_config) => Some(kafka_config.topic.clone()), - _ => None, + let topic = if config.transport.name() == TransportConfig::KAFKA_OUTPUT { + let kafka_config: feldera_types::transport::kafka::KafkaOutputConfig = + config.transport.deserialize_config().map_err(|e| { + ControllerError::encoder_config_parse_error( + endpoint_name, + &e, + &serde_json::to_string(config).unwrap_or_default(), + ) + })?; + Some(kafka_config.topic) + } else { + None }; if avro_config.threads == 0 { diff --git a/crates/adapters/src/format/csv.rs b/crates/adapters/src/format/csv.rs index 03d6fd01f54..ef0777cb81a 100644 --- a/crates/adapters/src/format/csv.rs +++ b/crates/adapters/src/format/csv.rs @@ -346,10 +346,7 @@ impl OutputFormat for CsvOutputFormat { ) })?; - if matches!( - config.transport, - feldera_types::config::TransportConfig::RedisOutput(_) - ) { + if config.transport.name() == feldera_types::config::TransportConfig::REDIS_OUTPUT { return Err(ControllerError::invalid_encoder_configuration( endpoint_name, "'csv' format not yet supported with Redis connector", diff --git a/crates/adapters/src/format/json/output.rs b/crates/adapters/src/format/json/output.rs index 3de37b66c1e..753cf55e490 100644 --- a/crates/adapters/src/format/json/output.rs +++ b/crates/adapters/src/format/json/output.rs @@ -74,7 +74,7 @@ impl OutputFormat for JsonOutputFormat { ) })?; - if matches!(&config.transport, TransportConfig::RedisOutput(_)) { + if config.transport.name() == TransportConfig::REDIS_OUTPUT { json_config.update_format = JsonUpdateFormat::Redis; }; @@ -88,9 +88,18 @@ impl OutputFormat for JsonOutputFormat { json_config.buffer_size_records = 1; } - let key_separator = match &config.transport { - TransportConfig::RedisOutput(config) => Some(config.key_separator.clone()), - _ => None, + let key_separator = if config.transport.name() == TransportConfig::REDIS_OUTPUT { + let redis_config: feldera_types::transport::redis::RedisOutputConfig = + config.transport.deserialize_config().map_err(|e| { + ControllerError::encoder_config_parse_error( + endpoint_name, + &e, + &serde_json::to_string(config).unwrap_or_default(), + ) + })?; + Some(redis_config.key_separator) + } else { + None }; Ok(Box::new(JsonEncoder::new( diff --git a/crates/adapters/src/format/parquet.rs b/crates/adapters/src/format/parquet.rs index 8c0fa93876e..76a2e0732bb 100644 --- a/crates/adapters/src/format/parquet.rs +++ b/crates/adapters/src/format/parquet.rs @@ -218,10 +218,7 @@ impl OutputFormat for ParquetOutputFormat { )); } - if matches!( - config.transport, - feldera_types::config::TransportConfig::RedisOutput(_) - ) { + if config.transport.name() == feldera_types::config::TransportConfig::REDIS_OUTPUT { return Err(ControllerError::invalid_encoder_configuration( endpoint_name, "'parquet' format not supported with Redis connector", diff --git a/crates/adapters/src/integrated.rs b/crates/adapters/src/integrated.rs index a36df0f8d3d..a5a2f48b651 100644 --- a/crates/adapters/src/integrated.rs +++ b/crates/adapters/src/integrated.rs @@ -4,6 +4,7 @@ use crate::{ControllerError, Encoder, InputConsumer, OutputEndpoint}; use datafusion::execution::runtime_env::RuntimeEnv; use feldera_types::config::{ConnectorConfig, PipelineConfig, TransportConfig}; use feldera_types::program_schema::Relation; +use serde::de::DeserializeOwned; use std::sync::{Arc, Weak}; #[cfg(feature = "with-deltalake")] @@ -27,6 +28,18 @@ pub trait IntegratedOutputEndpoint: OutputEndpoint + Encoder { fn as_endpoint(&mut self) -> &mut dyn OutputEndpoint; } +fn parse_transport_config( + endpoint_name: &str, + transport: &TransportConfig, +) -> Result +where + T: DeserializeOwned, +{ + transport.deserialize_config().map_err(|e| { + ControllerError::invalid_transport_configuration(endpoint_name, &e.to_string()) + }) +} + impl IntegratedOutputEndpoint for EP where EP: OutputEndpoint + Encoder + 'static, @@ -42,6 +55,8 @@ where /// Create an instance of an integrated output endpoint given its config /// and output relation schema. +/// Integrated connectors stay outside the transport registry because they +/// need schema and controller lifecycle state when they are constructed. #[allow(unused, clippy::too_many_arguments)] pub fn create_integrated_output_endpoint( endpoint_id: EndpointId, @@ -53,41 +68,41 @@ pub fn create_integrated_output_endpoint( continue_previous_state: bool, is_index: bool, ) -> Result, ControllerError> { - let ep: Box = match &connector_config.transport { + let ep: Box = match connector_config.transport.name() { #[cfg(feature = "with-deltalake")] - TransportConfig::DeltaTableOutput(config) => Box::new(delta_table::DeltaTableWriter::new( + TransportConfig::DELTA_TABLE_OUTPUT => Box::new(delta_table::DeltaTableWriter::new( endpoint_id, endpoint_name, - config, + &parse_transport_config(endpoint_name, &connector_config.transport)?, key_schema, schema, controller, continue_previous_state, is_index, )?), - TransportConfig::PostgresOutput(config) => Box::new(PostgresOutputEndpoint::new( + TransportConfig::POSTGRES_OUTPUT => Box::new(PostgresOutputEndpoint::new( endpoint_id, endpoint_name, - config, + &parse_transport_config(endpoint_name, &connector_config.transport)?, key_schema, schema, controller, is_index, )?), #[cfg(feature = "with-dynamodb")] - TransportConfig::DynamoDBOutput(config) => Box::new(DynamoDBOutputEndpoint::new( + TransportConfig::DYNAMODB_OUTPUT => Box::new(DynamoDBOutputEndpoint::new( endpoint_id, endpoint_name, - config, + &parse_transport_config(endpoint_name, &connector_config.transport)?, key_schema, schema, controller, is_index, )?), - transport => { + transport_name => { return Err(ControllerError::unknown_output_transport( endpoint_name, - &transport.name(), + transport_name, )); } }; @@ -113,40 +128,38 @@ pub fn create_integrated_input_endpoint( runtime_env: Arc, consumer: Box, ) -> Result, ControllerError> { - let ep: Box = match &config.transport { + let ep: Box = match config.transport.name() { #[cfg(feature = "with-deltalake")] - TransportConfig::DeltaTableInput(config) => { - Box::new(delta_table::DeltaTableInputEndpoint::new( - endpoint_name, - config, - pipeline_config, - runtime_env, - consumer, - )) - } + TransportConfig::DELTA_TABLE_INPUT => Box::new(delta_table::DeltaTableInputEndpoint::new( + endpoint_name, + &parse_transport_config(endpoint_name, &config.transport)?, + pipeline_config, + runtime_env, + consumer, + )), #[cfg(feature = "with-iceberg")] - TransportConfig::IcebergInput(config) => { - Box::new(feldera_iceberg::IcebergInputEndpoint::new( - endpoint_name, - config, - pipeline_config, - runtime_env, - consumer, - )) - } - TransportConfig::PostgresInput(config) => { - Box::new(PostgresInputEndpoint::new(endpoint_name, config, consumer)) - } + TransportConfig::ICEBERG_INPUT => Box::new(feldera_iceberg::IcebergInputEndpoint::new( + endpoint_name, + &parse_transport_config(endpoint_name, &config.transport)?, + pipeline_config, + runtime_env, + consumer, + )), + TransportConfig::POSTGRES_INPUT => Box::new(PostgresInputEndpoint::new( + endpoint_name, + &parse_transport_config(endpoint_name, &config.transport)?, + consumer, + )), #[cfg(feature = "with-postgres-cdc")] - TransportConfig::PostgresCdcInput(config) => Box::new(PostgresCdcInputEndpoint::new( + TransportConfig::POSTGRES_CDC_INPUT => Box::new(PostgresCdcInputEndpoint::new( endpoint_name, - config, + &parse_transport_config(endpoint_name, &config.transport)?, consumer, )), - transport => { + transport_name => { return Err(ControllerError::unknown_input_transport( endpoint_name, - &transport.name(), + transport_name, )); } }; diff --git a/crates/adapters/src/server.rs b/crates/adapters/src/server.rs index b0b18548ce1..798a695a009 100644 --- a/crates/adapters/src/server.rs +++ b/crates/adapters/src/server.rs @@ -2254,8 +2254,11 @@ async fn get_or_create_http_input_endpoint( // Create endpoint config. let config = InputEndpointConfig::new( table_name, - ConnectorConfig::new(TransportConfig::HttpInput(config), Some(format)) - .with_max_queued_records(HttpInputTransport::default_max_buffered_records()), + ConnectorConfig::new( + TransportConfig::new(TransportConfig::HTTP_INPUT, config), + Some(format), + ) + .with_max_queued_records(HttpInputTransport::default_max_buffered_records()), ); controller @@ -2436,8 +2439,11 @@ async fn output_endpoint( { object.insert( String::from("transport"), - serde_json::to_value(TransportConfig::HttpOutput(HttpOutputConfig::default())) - .unwrap(), + serde_json::to_value(TransportConfig::new( + TransportConfig::HTTP_OUTPUT, + HttpOutputConfig::default(), + )) + .unwrap(), ); } serde_json::from_value(json).map_err(|e| PipelineError::InvalidParam { @@ -2447,9 +2453,12 @@ async fn output_endpoint( let format = encoder_config_from_http_request(&table_name, args.format.unwrap_or_default(), &req)?; let mut connector_config = ConnectorConfig::new( - TransportConfig::HttpOutput(HttpOutputConfig { - backpressure: args.backpressure.unwrap_or_default(), - }), + TransportConfig::new( + TransportConfig::HTTP_OUTPUT, + HttpOutputConfig { + backpressure: args.backpressure.unwrap_or_default(), + }, + ), Some(format), ) .with_max_queued_records(HttpOutputTransport::default_max_buffered_records()); @@ -2458,14 +2467,20 @@ async fn output_endpoint( }; let config = OutputEndpointConfig::new(table_name, connector_config); - let http_output_config = match &config.connector_config.transport { - TransportConfig::HttpOutput(config) => config.clone(), - _ => { + let http_output_config = + if config.connector_config.transport.name() == TransportConfig::HTTP_OUTPUT { + config + .connector_config + .transport + .deserialize_config() + .map_err(|e| PipelineError::InvalidParam { + error: format!("Invalid HTTP output transport configuration: {e}"), + })? + } else { return Err(PipelineError::InvalidParam { error: "Transport configuration must be `http_output`".to_string(), }); - } - }; + }; let format = match &config.connector_config.format { Some(format) if format.name == "csv" => HttpOutputFormat::Csv, Some(format) if format.name == "json" => { diff --git a/crates/adapters/src/test/datagen.rs b/crates/adapters/src/test/datagen.rs index 574f991f09d..9368101d611 100644 --- a/crates/adapters/src/test/datagen.rs +++ b/crates/adapters/src/test/datagen.rs @@ -830,10 +830,17 @@ fn missing_config_does_something_sane() { })) .unwrap(); - if let TransportConfig::Datagen(dtg) = config.connector_config.transport { - assert_eq!(dtg.plan.len(), 1); - assert_eq!(dtg.plan[0], GenerationPlan::default()); - } + assert_eq!( + config.connector_config.transport.name(), + TransportConfig::DATAGEN + ); + let dtg: feldera_types::transport::datagen::DatagenInputConfig = config + .connector_config + .transport + .deserialize_config() + .unwrap(); + assert_eq!(dtg.plan.len(), 1); + assert_eq!(dtg.plan[0], GenerationPlan::default()); } #[test] diff --git a/crates/adapters/src/transport.rs b/crates/adapters/src/transport.rs index f634e86c0e4..ae67eea4801 100644 --- a/crates/adapters/src/transport.rs +++ b/crates/adapters/src/transport.rs @@ -80,6 +80,254 @@ use feldera_datagen::GeneratorEndpoint; pub use feldera_adapterlib::transport::*; +pub fn builtin_input_transport_registry() -> InputTransportRegistry { + let mut registry = InputTransportRegistry::new(); + registry.register(TransportConfig::FILE_INPUT, Box::new(FileInputFactory)); + #[cfg(feature = "with-kafka")] + registry.register(TransportConfig::KAFKA_INPUT, Box::new(KafkaInputFactory)); + #[cfg(feature = "with-nats")] + registry.register(TransportConfig::NATS_INPUT, Box::new(NatsInputFactory)); + #[cfg(feature = "with-pubsub")] + registry.register(TransportConfig::PUB_SUB_INPUT, Box::new(PubSubInputFactory)); + registry.register(TransportConfig::URL_INPUT, Box::new(UrlInputFactory)); + registry.register(TransportConfig::S3_INPUT, Box::new(S3InputFactory)); + registry.register(TransportConfig::DATAGEN, Box::new(DatagenInputFactory)); + #[cfg(feature = "with-nexmark")] + registry.register(TransportConfig::NEXMARK, Box::new(NexmarkInputFactory)); + registry.register(TransportConfig::HTTP_INPUT, Box::new(HttpInputFactory)); + registry.register(TransportConfig::ADHOC_INPUT, Box::new(AdHocInputFactory)); + registry.register(TransportConfig::CLOCK, Box::new(ClockInputFactory)); + registry.register(TransportConfig::EMPTY_INPUT, Box::new(EmptyInputFactory)); + registry +} + +pub fn builtin_output_transport_registry() -> OutputTransportRegistry { + let mut registry = OutputTransportRegistry::new(); + registry.register(TransportConfig::FILE_OUTPUT, Box::new(FileOutputFactory)); + #[cfg(feature = "with-kafka")] + registry.register(TransportConfig::KAFKA_OUTPUT, Box::new(KafkaOutputFactory)); + #[cfg(feature = "with-redis")] + registry.register(TransportConfig::REDIS_OUTPUT, Box::new(RedisOutputFactory)); + registry.register(TransportConfig::NULL_OUTPUT, Box::new(NullOutputFactory)); + registry +} + +struct FileInputFactory; + +impl InputTransportEndpointFactory for FileInputFactory { + fn create( + &self, + config: &TransportConfig, + ) -> AnyResult>> { + let config = config.deserialize_config()?; + Ok(Some(Box::new(FileInputEndpoint::new(config)))) + } +} + +#[cfg(feature = "with-kafka")] +struct KafkaInputFactory; + +#[cfg(feature = "with-kafka")] +impl InputTransportEndpointFactory for KafkaInputFactory { + fn create( + &self, + config: &TransportConfig, + ) -> AnyResult>> { + let config = config.deserialize_config()?; + Ok(Some(Box::new(KafkaFtInputEndpoint::new(config)?))) + } +} + +#[cfg(feature = "with-nats")] +struct NatsInputFactory; + +#[cfg(feature = "with-nats")] +impl InputTransportEndpointFactory for NatsInputFactory { + fn create( + &self, + config: &TransportConfig, + ) -> AnyResult>> { + let config = config.deserialize_config()?; + Ok(Some(Box::new(NatsInputEndpoint::new(config)?))) + } +} + +#[cfg(feature = "with-pubsub")] +struct PubSubInputFactory; + +#[cfg(feature = "with-pubsub")] +impl InputTransportEndpointFactory for PubSubInputFactory { + fn create( + &self, + config: &TransportConfig, + ) -> AnyResult>> { + let config = config.deserialize_config()?; + Ok(Some(Box::new(PubSubInputEndpoint::new(config)?))) + } +} + +struct UrlInputFactory; + +impl InputTransportEndpointFactory for UrlInputFactory { + fn create( + &self, + config: &TransportConfig, + ) -> AnyResult>> { + let config = config.deserialize_config()?; + Ok(Some(Box::new(UrlInputEndpoint::new(config)))) + } +} + +struct S3InputFactory; + +impl InputTransportEndpointFactory for S3InputFactory { + fn create( + &self, + config: &TransportConfig, + ) -> AnyResult>> { + let config = config.deserialize_config()?; + Ok(Some(Box::new(S3InputEndpoint::new(config)?))) + } +} + +struct DatagenInputFactory; + +impl InputTransportEndpointFactory for DatagenInputFactory { + fn create( + &self, + config: &TransportConfig, + ) -> AnyResult>> { + let config = config.deserialize_config()?; + Ok(Some(Box::new(GeneratorEndpoint::new(config)))) + } +} + +#[cfg(feature = "with-nexmark")] +struct NexmarkInputFactory; + +#[cfg(feature = "with-nexmark")] +impl InputTransportEndpointFactory for NexmarkInputFactory { + fn create( + &self, + config: &TransportConfig, + ) -> AnyResult>> { + let config = config.deserialize_config()?; + Ok(Some(Box::new(NexmarkEndpoint::new(config)))) + } +} + +struct HttpInputFactory; + +impl InputTransportEndpointFactory for HttpInputFactory { + fn create( + &self, + config: &TransportConfig, + ) -> AnyResult>> { + let config = config.deserialize_config()?; + Ok(Some(Box::new(HttpInputEndpoint::new(config)))) + } +} + +struct AdHocInputFactory; + +impl InputTransportEndpointFactory for AdHocInputFactory { + fn create( + &self, + config: &TransportConfig, + ) -> AnyResult>> { + let config = config.deserialize_config()?; + Ok(Some(Box::new(AdHocInputEndpoint::new(config)))) + } +} + +struct ClockInputFactory; + +impl InputTransportEndpointFactory for ClockInputFactory { + fn create( + &self, + config: &TransportConfig, + ) -> AnyResult>> { + let config = config.deserialize_config()?; + Ok(Some(Box::new(ClockEndpoint::new(config)?))) + } +} + +struct EmptyInputFactory; + +impl InputTransportEndpointFactory for EmptyInputFactory { + fn create( + &self, + config: &TransportConfig, + ) -> AnyResult>> { + Ok(Some(Box::new(EmptyInputEndpoint))) + } +} + +struct FileOutputFactory; + +impl OutputTransportEndpointFactory for FileOutputFactory { + fn create( + &self, + config: &TransportConfig, + _endpoint_name: &str, + _fault_tolerant: bool, + ) -> AnyResult>> { + let config = config.deserialize_config()?; + Ok(Some(Box::new(FileOutputEndpoint::new(config)?))) + } +} + +#[cfg(feature = "with-kafka")] +struct KafkaOutputFactory; + +#[cfg(feature = "with-kafka")] +impl OutputTransportEndpointFactory for KafkaOutputFactory { + fn create( + &self, + config: &TransportConfig, + endpoint_name: &str, + fault_tolerant: bool, + ) -> AnyResult>> { + let config = config.deserialize_config()?; + match fault_tolerant { + false => Ok(Some(Box::new(KafkaOutputEndpoint::new( + config, + endpoint_name, + )?))), + true => Ok(Some(Box::new(KafkaFtOutputEndpoint::new(config)?))), + } + } +} + +#[cfg(feature = "with-redis")] +struct RedisOutputFactory; + +#[cfg(feature = "with-redis")] +impl OutputTransportEndpointFactory for RedisOutputFactory { + fn create( + &self, + config: &TransportConfig, + _endpoint_name: &str, + _fault_tolerant: bool, + ) -> AnyResult>> { + let config = config.deserialize_config()?; + Ok(Some(Box::new(RedisOutputEndpoint::new(config)?))) + } +} + +struct NullOutputFactory; + +impl OutputTransportEndpointFactory for NullOutputFactory { + fn create( + &self, + config: &TransportConfig, + _endpoint_name: &str, + _fault_tolerant: bool, + ) -> AnyResult>> { + Ok(Some(Box::new(NullOutputEndpoint))) + } +} + /// Creates an input transport endpoint instance using an input transport /// configuration, resolving secrets by reading `secrets_dir`. /// @@ -90,47 +338,19 @@ pub fn input_transport_config_to_endpoint( config: &TransportConfig, endpoint_name: &str, secrets_dir: &Path, +) -> AnyResult>> { + let registry = builtin_input_transport_registry(); + input_transport_config_to_endpoint_with_registry(config, endpoint_name, secrets_dir, ®istry) +} + +fn input_transport_config_to_endpoint_with_registry( + config: &TransportConfig, + _endpoint_name: &str, + secrets_dir: &Path, + registry: &InputTransportRegistry, ) -> AnyResult>> { let config = resolve_secret_references_via_json(secrets_dir, config)?; - let endpoint: Box = match config { - TransportConfig::FileInput(config) => Box::new(FileInputEndpoint::new(config)), - #[cfg(feature = "with-kafka")] - TransportConfig::KafkaInput(config) => Box::new(KafkaFtInputEndpoint::new(config)?), - #[cfg(not(feature = "with-kafka"))] - TransportConfig::KafkaInput(_) => return Ok(None), - #[cfg(feature = "with-nats")] - TransportConfig::NatsInput(config) => Box::new(NatsInputEndpoint::new(config)?), - #[cfg(not(feature = "with-nats"))] - TransportConfig::NatsInput(_) => return Ok(None), - #[cfg(feature = "with-pubsub")] - TransportConfig::PubSubInput(config) => Box::new(PubSubInputEndpoint::new(config.clone())?), - #[cfg(not(feature = "with-pubsub"))] - TransportConfig::PubSubInput(_) => return Ok(None), - TransportConfig::UrlInput(config) => Box::new(UrlInputEndpoint::new(config)), - TransportConfig::S3Input(config) => Box::new(S3InputEndpoint::new(config)?), - TransportConfig::Datagen(config) => Box::new(GeneratorEndpoint::new(config.clone())), - #[cfg(feature = "with-nexmark")] - TransportConfig::Nexmark(config) => Box::new(NexmarkEndpoint::new(config.clone())), - #[cfg(not(feature = "with-nexmark"))] - TransportConfig::Nexmark(_) => return Ok(None), - TransportConfig::HttpInput(config) => Box::new(HttpInputEndpoint::new(config)), - TransportConfig::AdHocInput(config) => Box::new(AdHocInputEndpoint::new(config)), - TransportConfig::ClockInput(config) => Box::new(ClockEndpoint::new(config)?), - TransportConfig::EmptyInput => Box::new(EmptyInputEndpoint), - TransportConfig::FileOutput(_) - | TransportConfig::KafkaOutput(_) - | TransportConfig::DeltaTableInput(_) - | TransportConfig::DeltaTableOutput(_) - | TransportConfig::DynamoDBOutput(_) - | TransportConfig::PostgresInput(_) - | TransportConfig::PostgresCdcInput(_) - | TransportConfig::PostgresOutput(_) - | TransportConfig::HttpOutput(_) - | TransportConfig::RedisOutput(_) - | TransportConfig::IcebergInput(_) - | TransportConfig::NullOutput => return Ok(None), - }; - Ok(Some(endpoint)) + registry.create_endpoint(&config) } /// Creates an output transport endpoint instance using an output transport @@ -148,23 +368,259 @@ pub fn output_transport_config_to_endpoint( endpoint_name: &str, fault_tolerant: bool, secrets_dir: &Path, +) -> AnyResult>> { + let registry = builtin_output_transport_registry(); + output_transport_config_to_endpoint_with_registry( + config, + endpoint_name, + fault_tolerant, + secrets_dir, + ®istry, + ) +} + +fn output_transport_config_to_endpoint_with_registry( + config: &TransportConfig, + endpoint_name: &str, + fault_tolerant: bool, + secrets_dir: &Path, + registry: &OutputTransportRegistry, ) -> AnyResult>> { let config = resolve_secret_references_via_json(secrets_dir, config)?; - match config { - TransportConfig::FileOutput(config) => Ok(Some(Box::new(FileOutputEndpoint::new(config)?))), + registry.create_endpoint(&config, endpoint_name, fault_tolerant) +} + +#[cfg(test)] +mod tests { + use super::*; + use feldera_types::config::FtModel; + use serde_json::json; + use std::fs::{File, create_dir_all}; + use std::io::Write; + use std::path::Path; + + struct SecretAssertingInputFactory; + + impl InputTransportEndpointFactory for SecretAssertingInputFactory { + fn create( + &self, + config: &TransportConfig, + ) -> AnyResult>> { + assert_eq!(config.config["path"], json!("resolved-input-path")); + Ok(None) + } + } + + struct SecretAssertingOutputFactory; + + impl OutputTransportEndpointFactory for SecretAssertingOutputFactory { + fn create( + &self, + config: &TransportConfig, + _endpoint_name: &str, + _fault_tolerant: bool, + ) -> AnyResult>> { + assert_eq!(config.config["path"], json!("resolved-output-path")); + Ok(None) + } + } + + fn write_test_secret(secrets_dir: &Path, key: &str, value: &str) { + let name_dir = secrets_dir.join("kubernetes").join("transport"); + create_dir_all(&name_dir).unwrap(); + let mut file = File::create(name_dir.join(key)).unwrap(); + file.write_all(value.as_bytes()).unwrap(); + } + + #[test] + fn builtin_input_registry_creates_empty_input_endpoint() { + let secrets_dir = tempfile::tempdir().unwrap(); + + let endpoint = input_transport_config_to_endpoint( + &TransportConfig::without_config(TransportConfig::EMPTY_INPUT), + "empty", + secrets_dir.path(), + ) + .unwrap() + .unwrap(); + + assert_eq!(endpoint.fault_tolerance(), Some(FtModel::ExactlyOnce)); + } + + #[test] + fn builtin_output_registry_creates_null_output_endpoint() { + let secrets_dir = tempfile::tempdir().unwrap(); + + let endpoint = output_transport_config_to_endpoint( + &TransportConfig::without_config(TransportConfig::NULL_OUTPUT), + "null", + true, + secrets_dir.path(), + ) + .unwrap() + .unwrap(); + + assert!(endpoint.is_fault_tolerant()); + assert_eq!(endpoint.max_buffer_size_bytes(), usize::MAX); + } + + #[test] + fn wrong_direction_transport_configs_still_return_none() { + let secrets_dir = tempfile::tempdir().unwrap(); + + assert!( + input_transport_config_to_endpoint( + &TransportConfig::without_config(TransportConfig::NULL_OUTPUT), + "null", + secrets_dir.path() + ) + .unwrap() + .is_none() + ); + assert!( + output_transport_config_to_endpoint( + &TransportConfig::without_config(TransportConfig::EMPTY_INPUT), + "empty", + false, + secrets_dir.path() + ) + .unwrap() + .is_none() + ); + } + + #[test] + fn explicit_transport_registries_dispatch_by_transport_name() { + let mut input_registry = InputTransportRegistry::new(); + assert!( + input_registry + .create_endpoint(&TransportConfig::without_config( + TransportConfig::EMPTY_INPUT + )) + .unwrap() + .is_none() + ); + input_registry.register(TransportConfig::EMPTY_INPUT, Box::new(EmptyInputFactory)); + assert!( + input_registry + .create_endpoint(&TransportConfig::without_config( + TransportConfig::EMPTY_INPUT + )) + .unwrap() + .is_some() + ); + input_registry.register(String::from("dynamic_input"), Box::new(EmptyInputFactory)); + assert!(input_registry.get("dynamic_input").is_some()); + + let mut output_registry = OutputTransportRegistry::new(); + assert!( + output_registry + .create_endpoint( + &TransportConfig::without_config(TransportConfig::NULL_OUTPUT), + "null", + true + ) + .unwrap() + .is_none() + ); + output_registry.register(TransportConfig::NULL_OUTPUT, Box::new(NullOutputFactory)); + assert!( + output_registry + .create_endpoint( + &TransportConfig::without_config(TransportConfig::NULL_OUTPUT), + "null", + true + ) + .unwrap() + .is_some() + ); + output_registry.register(String::from("dynamic_output"), Box::new(NullOutputFactory)); + assert!(output_registry.get("dynamic_output").is_some()); + } + + #[test] + fn input_transport_secrets_are_resolved_before_factory_parsing() { + let secrets_dir = tempfile::tempdir().unwrap(); + write_test_secret(secrets_dir.path(), "input", "resolved-input-path"); + + let config = TransportConfig::new( + "secret_asserting_input", + json!({"path": "${secret:kubernetes:transport/input}"}), + ); + + let mut input_registry = InputTransportRegistry::new(); + input_registry.register( + "secret_asserting_input", + Box::new(SecretAssertingInputFactory), + ); + + assert!( + input_transport_config_to_endpoint_with_registry( + &config, + "input", + secrets_dir.path(), + &input_registry, + ) + .unwrap() + .is_none() + ) + } + + #[test] + fn output_transport_secrets_are_resolved_before_factory_parsing() { + let secrets_dir = tempfile::tempdir().unwrap(); + write_test_secret(secrets_dir.path(), "output", "resolved-output-path"); + + let config = TransportConfig::new( + "secret_asserting_output", + json!({"path": "${secret:kubernetes:transport/output}"}), + ); + + let mut output_registry = OutputTransportRegistry::new(); + output_registry.register( + "secret_asserting_output", + Box::new(SecretAssertingOutputFactory), + ); + + assert!( + output_transport_config_to_endpoint_with_registry( + &config, + "output", + false, + secrets_dir.path(), + &output_registry, + ) + .unwrap() + .is_none() + ) + } + + #[test] + fn builtin_transport_registries_include_compiled_transport_names() { + let input_registry = builtin_input_transport_registry(); + assert!(input_registry.get(TransportConfig::FILE_INPUT).is_some()); #[cfg(feature = "with-kafka")] - TransportConfig::KafkaOutput(config) => match fault_tolerant { - false => Ok(Some(Box::new(KafkaOutputEndpoint::new( - config, - endpoint_name, - )?))), - true => Ok(Some(Box::new(KafkaFtOutputEndpoint::new(config)?))), - }, + assert!(input_registry.get(TransportConfig::KAFKA_INPUT).is_some()); + #[cfg(feature = "with-nats")] + assert!(input_registry.get(TransportConfig::NATS_INPUT).is_some()); + #[cfg(feature = "with-pubsub")] + assert!(input_registry.get(TransportConfig::PUB_SUB_INPUT).is_some()); + assert!(input_registry.get(TransportConfig::URL_INPUT).is_some()); + assert!(input_registry.get(TransportConfig::S3_INPUT).is_some()); + assert!(input_registry.get(TransportConfig::DATAGEN).is_some()); + #[cfg(feature = "with-nexmark")] + assert!(input_registry.get(TransportConfig::NEXMARK).is_some()); + assert!(input_registry.get(TransportConfig::HTTP_INPUT).is_some()); + assert!(input_registry.get(TransportConfig::ADHOC_INPUT).is_some()); + assert!(input_registry.get(TransportConfig::CLOCK).is_some()); + assert!(input_registry.get(TransportConfig::EMPTY_INPUT).is_some()); + + let output_registry = builtin_output_transport_registry(); + assert!(output_registry.get(TransportConfig::FILE_OUTPUT).is_some()); + #[cfg(feature = "with-kafka")] + assert!(output_registry.get(TransportConfig::KAFKA_OUTPUT).is_some()); #[cfg(feature = "with-redis")] - TransportConfig::RedisOutput(config) => { - Ok(Some(Box::new(RedisOutputEndpoint::new(config)?))) - } - TransportConfig::NullOutput => Ok(Some(Box::new(NullOutputEndpoint))), - _ => Ok(None), + assert!(output_registry.get(TransportConfig::REDIS_OUTPUT).is_some()); + assert!(output_registry.get(TransportConfig::NULL_OUTPUT).is_some()); } } diff --git a/crates/adapters/src/transport/clock.rs b/crates/adapters/src/transport/clock.rs index 579c9dec872..dbc80a1148d 100644 --- a/crates/adapters/src/transport/clock.rs +++ b/crates/adapters/src/transport/clock.rs @@ -67,14 +67,17 @@ pub fn now_endpoint_config(config: &PipelineConfig) -> InputEndpointConfig { InputEndpointConfig::new( "now", ConnectorConfig::new( - TransportConfig::ClockInput(ClockConfig { - clock_resolution_usecs: config - .global - .clock_resolution_usecs - .unwrap_or(DEFAULT_CLOCK_RESOLUTION_USECS), - now_offset_ms: config.global.dev_tweaks.now_offset_ms(), - http_driven: config.global.dev_tweaks.now_http_driven(), - }), + TransportConfig::new( + TransportConfig::CLOCK, + ClockConfig { + clock_resolution_usecs: config + .global + .clock_resolution_usecs + .unwrap_or(DEFAULT_CLOCK_RESOLUTION_USECS), + now_offset_ms: config.global.dev_tweaks.now_offset_ms(), + http_driven: config.global.dev_tweaks.now_http_driven(), + }, + ), Some(FormatConfig { name: Cow::Borrowed("json"), config: serde_json::to_value(JsonParserConfig { @@ -722,10 +725,15 @@ mod test { let target_of = |body: serde_json::Value| -> Option { let config: PipelineConfig = serde_json::from_value(body).unwrap(); let endpoint = super::now_endpoint_config(&config); - let TransportConfig::ClockInput(clock_config) = &endpoint.connector_config.transport - else { - panic!("expected ClockInput transport"); - }; + assert_eq!( + endpoint.connector_config.transport.name(), + TransportConfig::CLOCK + ); + let clock_config: ClockConfig = endpoint + .connector_config + .transport + .deserialize_config() + .unwrap(); clock_config.now_offset_ms }; diff --git a/crates/adapters/src/transport/kafka/ft/test.rs b/crates/adapters/src/transport/kafka/ft/test.rs index d32e7c56ce7..adb7335c8f7 100644 --- a/crates/adapters/src/transport/kafka/ft/test.rs +++ b/crates/adapters/src/transport/kafka/ft/test.rs @@ -1540,24 +1540,28 @@ fn test_offset( let config = InputEndpointConfig::new( "test_input", ConnectorConfig::new( - TransportConfig::KafkaInput(KafkaInputConfig { - log_level: Some(KafkaLogLevel::Debug), - start_from: match start_from { - KafkaStartFromConfig::Timestamp(_) => { - sleep(Duration::from_secs(2)); - let timestamp = - KafkaStartFromConfig::Timestamp(Timestamp::now().to_millis().unwrap()); - sleep(Duration::from_secs(2)); - if let Some(send_after) = send_after.take() { - producer.send_to_topic(send_after, topic); - producer.send_string("", topic); + TransportConfig::new( + TransportConfig::KAFKA_INPUT, + KafkaInputConfig { + log_level: Some(KafkaLogLevel::Debug), + start_from: match start_from { + KafkaStartFromConfig::Timestamp(_) => { + sleep(Duration::from_secs(2)); + let timestamp = KafkaStartFromConfig::Timestamp( + Timestamp::now().to_millis().unwrap(), + ); + sleep(Duration::from_secs(2)); + if let Some(send_after) = send_after.take() { + producer.send_to_topic(send_after, topic); + producer.send_string("", topic); + } + timestamp } - timestamp - } - other => other, + other => other, + }, + ..KafkaInputConfig::default(kafka_options, topic) }, - ..KafkaInputConfig::default(kafka_options, topic) - }), + ), Some(FormatConfig { name: Cow::from("csv"), config: json!({}), @@ -1971,12 +1975,15 @@ fn test_input_partition( let config = InputEndpointConfig::new( "test_input", ConnectorConfig::new( - TransportConfig::KafkaInput(KafkaInputConfig { - log_level: Some(KafkaLogLevel::Debug), - start_from: start_from.clone(), - partitions: Some(partitions.clone()), - ..KafkaInputConfig::default(kafka_options, topic) - }), + TransportConfig::new( + TransportConfig::KAFKA_INPUT, + KafkaInputConfig { + log_level: Some(KafkaLogLevel::Debug), + start_from: start_from.clone(), + partitions: Some(partitions.clone()), + ..KafkaInputConfig::default(kafka_options, topic) + }, + ), Some(FormatConfig { name: Cow::from("csv"), config: json!({}), diff --git a/crates/adapters/src/transport/s3.rs b/crates/adapters/src/transport/s3.rs index 62142bc63a7..e3ca748123c 100644 --- a/crates/adapters/src/transport/s3.rs +++ b/crates/adapters/src/transport/s3.rs @@ -1038,12 +1038,9 @@ mod test { ) { let config: InputEndpointConfig = serde_json::from_str(config_str).unwrap(); let transport_config = config.connector_config.transport.clone(); - let transport_config: Arc = match transport_config { - TransportConfig::S3Input(config) => Arc::new(config), - _ => { - panic!("Expected S3Input transport configuration"); - } - }; + assert_eq!(transport_config.name(), TransportConfig::S3_INPUT); + let transport_config: Arc = + Arc::new(transport_config.deserialize_config().unwrap()); let (consumer, parser, input_handle) = mock_parser_pipeline::( &Relation::empty(), &config.connector_config.format.unwrap(), diff --git a/crates/feldera-types/src/config.rs b/crates/feldera-types/src/config.rs index ad34a38e74d..cb8bf875268 100644 --- a/crates/feldera-types/src/config.rs +++ b/crates/feldera-types/src/config.rs @@ -8,24 +8,6 @@ use crate::postprocess::PostprocessorConfig; use crate::preprocess::PreprocessorConfig; use crate::secret_resolver::default_secrets_directory; -use crate::transport::adhoc::AdHocInputConfig; -use crate::transport::clock::ClockConfig; -use crate::transport::datagen::DatagenInputConfig; -use crate::transport::delta_table::{DeltaTableReaderConfig, DeltaTableWriterConfig}; -use crate::transport::dynamodb::DynamoDBWriterConfig; -use crate::transport::file::{FileInputConfig, FileOutputConfig}; -use crate::transport::http::{HttpInputConfig, HttpOutputConfig}; -use crate::transport::iceberg::IcebergReaderConfig; -use crate::transport::kafka::{KafkaInputConfig, KafkaOutputConfig}; -use crate::transport::nats::NatsInputConfig; -use crate::transport::nexmark::NexmarkInputConfig; -use crate::transport::postgres::{ - PostgresCdcReaderConfig, PostgresReaderConfig, PostgresWriterConfig, -}; -use crate::transport::pubsub::PubSubInputConfig; -use crate::transport::redis::RedisOutputConfig; -use crate::transport::s3::S3InputConfig; -use crate::transport::url::UrlInputConfig; use core::fmt; use feldera_ir::{MirNode, MirNodeId}; use serde::de::{self, MapAccess, Visitor}; @@ -1210,8 +1192,11 @@ mod test { use super::deserialize_fault_tolerance; use crate::config::{ DEFAULT_DATAFUSION_MEMORY_MB_CEILING, FtConfig, FtModel, ResourceConfig, RuntimeConfig, + TransportConfig, }; + use crate::transport::datagen::DatagenInputConfig; use serde::{Deserialize, Serialize}; + use serde_json::json; #[test] fn resolved_datafusion_memory_explicit_passes_through() { @@ -1339,6 +1324,42 @@ mod test { } ); } + + #[test] + fn transport_config_deserializes_existing_config_payload_shape() { + let transport: TransportConfig = serde_json::from_value(json!({ + "name": "datagen", + "config": { + "workers": 3 + } + })) + .unwrap(); + + assert_eq!(transport.name(), TransportConfig::DATAGEN); + assert_eq!(transport.config, json!({"workers": 3})); + let typed_config: DatagenInputConfig = transport.deserialize_config().unwrap(); + assert_eq!(typed_config.workers, 3); + } + + #[test] + fn transport_config_deserializes_existing_unit_variant_shape() { + let transport: TransportConfig = + serde_json::from_value(json!({"name": "null_output"})).unwrap(); + + assert_eq!(transport.name(), TransportConfig::NULL_OUTPUT); + assert!(transport.config.is_null()); + } + + #[test] + fn transport_config_serializes_unit_transports_without_config_field() { + assert_eq!( + serde_json::to_value(TransportConfig::without_config( + TransportConfig::EMPTY_INPUT + )) + .unwrap(), + json!({"name": "empty_input"}) + ); + } } impl FtConfig { @@ -1819,91 +1840,89 @@ impl OutputEndpointConfig { } } -/// Transport-specific endpoint configuration passed to -/// `crate::OutputTransport::new_endpoint` -/// and `crate::InputTransport::new_endpoint`. +/// Transport-specific endpoint configuration. +/// +/// The `name` field selects a transport factory. The `config` field is an +/// opaque transport-specific JSON payload parsed by that factory. #[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize, ToSchema)] -#[serde(tag = "name", content = "config", rename_all = "snake_case")] -pub enum TransportConfig { - FileInput(FileInputConfig), - FileOutput(FileOutputConfig), - NatsInput(NatsInputConfig), - KafkaInput(KafkaInputConfig), - KafkaOutput(KafkaOutputConfig), - PubSubInput(PubSubInputConfig), - UrlInput(UrlInputConfig), - S3Input(S3InputConfig), - DeltaTableInput(DeltaTableReaderConfig), - DeltaTableOutput(DeltaTableWriterConfig), - // Snake case would rename "DynamoDBOutput" to `dynamo_db_output`. - // However, DynamoDB is a single word, so override the tag to `dynamodb_output`. - #[serde(rename = "dynamodb_output")] - DynamoDBOutput(DynamoDBWriterConfig), - RedisOutput(RedisOutputConfig), - // Prevent rust from complaining about large size difference between enum variants. - IcebergInput(Box), - PostgresInput(PostgresReaderConfig), - PostgresCdcInput(PostgresCdcReaderConfig), - PostgresOutput(PostgresWriterConfig), - Datagen(DatagenInputConfig), - Nexmark(NexmarkInputConfig), - /// Direct HTTP input: cannot be instantiated through API - HttpInput(HttpInputConfig), - /// Direct HTTP output: cannot be instantiated through API - HttpOutput(HttpOutputConfig), - /// Ad hoc input: cannot be instantiated through API - AdHocInput(AdHocInputConfig), - ClockInput(ClockConfig), - /// Output connector that discards all data. - NullOutput, - /// Input connector that produces no data. - EmptyInput, +pub struct TransportConfig { + /// Transport name, e.g. "kafka_input", "file_output", etc. + pub name: Cow<'static, str>, + + /// Transport-specific configuration. + #[serde(default, skip_serializing_if = "JsonValue::is_null")] + #[schema(value_type = Object)] + pub config: JsonValue, } impl TransportConfig { - pub fn name(&self) -> String { - match self { - TransportConfig::FileInput(_) => "file_input".to_string(), - TransportConfig::FileOutput(_) => "file_output".to_string(), - TransportConfig::NatsInput(_) => "nats_input".to_string(), - TransportConfig::KafkaInput(_) => "kafka_input".to_string(), - TransportConfig::KafkaOutput(_) => "kafka_output".to_string(), - TransportConfig::PubSubInput(_) => "pub_sub_input".to_string(), - TransportConfig::UrlInput(_) => "url_input".to_string(), - TransportConfig::S3Input(_) => "s3_input".to_string(), - TransportConfig::DeltaTableInput(_) => "delta_table_input".to_string(), - TransportConfig::DeltaTableOutput(_) => "delta_table_output".to_string(), - TransportConfig::DynamoDBOutput(_) => "dynamodb_output".to_string(), - TransportConfig::IcebergInput(_) => "iceberg_input".to_string(), - TransportConfig::PostgresInput(_) => "postgres_input".to_string(), - TransportConfig::PostgresCdcInput(_) => "postgres_cdc_input".to_string(), - TransportConfig::PostgresOutput(_) => "postgres_output".to_string(), - TransportConfig::Datagen(_) => "datagen".to_string(), - TransportConfig::Nexmark(_) => "nexmark".to_string(), - TransportConfig::HttpInput(_) => "http_input".to_string(), - TransportConfig::HttpOutput(_) => "http_output".to_string(), - TransportConfig::AdHocInput(_) => "adhoc_input".to_string(), - TransportConfig::RedisOutput(_) => "redis_output".to_string(), - TransportConfig::ClockInput(_) => "clock".to_string(), - TransportConfig::NullOutput => "null_output".to_string(), - TransportConfig::EmptyInput => "empty_input".to_string(), + pub const FILE_INPUT: &'static str = "file_input"; + pub const FILE_OUTPUT: &'static str = "file_output"; + pub const NATS_INPUT: &'static str = "nats_input"; + pub const KAFKA_INPUT: &'static str = "kafka_input"; + pub const KAFKA_OUTPUT: &'static str = "kafka_output"; + pub const PUB_SUB_INPUT: &'static str = "pub_sub_input"; + pub const URL_INPUT: &'static str = "url_input"; + pub const S3_INPUT: &'static str = "s3_input"; + pub const DELTA_TABLE_INPUT: &'static str = "delta_table_input"; + pub const DELTA_TABLE_OUTPUT: &'static str = "delta_table_output"; + pub const DYNAMODB_OUTPUT: &'static str = "dynamodb_output"; + pub const REDIS_OUTPUT: &'static str = "redis_output"; + pub const ICEBERG_INPUT: &'static str = "iceberg_input"; + pub const POSTGRES_INPUT: &'static str = "postgres_input"; + pub const POSTGRES_CDC_INPUT: &'static str = "postgres_cdc_input"; + pub const POSTGRES_OUTPUT: &'static str = "postgres_output"; + pub const DATAGEN: &'static str = "datagen"; + pub const NEXMARK: &'static str = "nexmark"; + pub const HTTP_INPUT: &'static str = "http_input"; + pub const HTTP_OUTPUT: &'static str = "http_output"; + pub const ADHOC_INPUT: &'static str = "adhoc_input"; + pub const CLOCK: &'static str = "clock"; + pub const NULL_OUTPUT: &'static str = "null_output"; + pub const EMPTY_INPUT: &'static str = "empty_input"; + + pub fn new(name: impl Into>, config: impl Serialize) -> Self { + Self { + name: name.into(), + config: serde_json::to_value(config) + .expect("transport configuration should be JSON-serializable"), + } + } + + pub fn without_config(name: impl Into>) -> Self { + Self { + name: name.into(), + config: JsonValue::Null, } } + pub fn name(&self) -> &str { + self.name.as_ref() + } + + pub fn deserialize_config(&self) -> Result + where + T: for<'de> Deserialize<'de>, + { + let config = if self.config.is_null() { + JsonValue::Object(Default::default()) + } else { + self.config.clone() + }; + serde_json::from_value(config) + } + /// Returns true if the connector is transient, i.e., is created and destroyed /// at runtime on demand, rather than being configured as part of the pipeline. pub fn is_transient(&self) -> bool { matches!( - self, - TransportConfig::AdHocInput(_) - | TransportConfig::HttpInput(_) - | TransportConfig::HttpOutput(_) - | TransportConfig::ClockInput(_) + self.name(), + Self::ADHOC_INPUT | Self::HTTP_INPUT | Self::HTTP_OUTPUT | Self::CLOCK ) } pub fn is_http_input(&self) -> bool { - matches!(self, TransportConfig::HttpInput(_)) + self.name() == Self::HTTP_INPUT } } diff --git a/crates/feldera-types/src/secret_resolver.rs b/crates/feldera-types/src/secret_resolver.rs index 66cf07d9870..03ba6eeeb1b 100644 --- a/crates/feldera-types/src/secret_resolver.rs +++ b/crates/feldera-types/src/secret_resolver.rs @@ -254,7 +254,7 @@ fn resolve_potential_secret_reference_string( #[cfg(test)] mod tests { - use crate::config::{ConnectorConfig, TransportConfig}; + use crate::config::ConnectorConfig; use crate::secret_ref::{MaybeSecretRef, SecretRef}; use crate::secret_resolver::{ SecretRefResolutionError, discover_secret_references_in_connector_config, @@ -562,17 +562,9 @@ mod tests { resolve_secret_references_in_connector_config(dir_path, &connector_config).unwrap(); // Transport configuration resolution - let TransportConfig::Datagen(datagen_input_config) = - connector_config_secrets_resolved.transport - else { - unreachable!(); - }; assert_eq!( - datagen_input_config.plan[0].fields["col2"] - .values - .as_ref() - .unwrap(), - &vec![json!("example1"), json!("example2")] + connector_config_secrets_resolved.transport.config["plan"][0]["fields"]["col2"]["values"], + json!(["example1", "example2"]) ); // Format configuration resolution @@ -669,15 +661,9 @@ mod tests { let resolved = resolve_secret_references_in_connector_config(dir.path(), &connector_config).unwrap(); - let TransportConfig::Datagen(datagen_input_config) = resolved.transport else { - unreachable!(); - }; assert_eq!( - datagen_input_config.plan[0].fields["col2"] - .values - .as_ref() - .unwrap(), - &vec![json!("resolved_value_a"), json!("resolved_value_b")] + resolved.transport.config["plan"][0]["fields"]["col2"]["values"], + json!(["resolved_value_a", "resolved_value_b"]) ); let Some(format_config) = resolved.format else { diff --git a/crates/pipeline-manager/src/compiler/sql_compiler.rs b/crates/pipeline-manager/src/compiler/sql_compiler.rs index a4f81fcd094..faad124e73d 100644 --- a/crates/pipeline-manager/src/compiler/sql_compiler.rs +++ b/crates/pipeline-manager/src/compiler/sql_compiler.rs @@ -1279,10 +1279,7 @@ mod test { .unwrap() .connector_config .clone(); - assert!(matches!( - connector_config.transport, - TransportConfig::Datagen(_) - )); + assert_eq!(connector_config.transport.name(), TransportConfig::DATAGEN); // Program schema only with properties: check properties let program_schema_properties_only: ProgramSchemaPropertiesOnly = diff --git a/crates/pipeline-manager/src/db/types/program.rs b/crates/pipeline-manager/src/db/types/program.rs index 4057794b639..9e4bbcefc7f 100644 --- a/crates/pipeline-manager/src/db/types/program.rs +++ b/crates/pipeline-manager/src/db/types/program.rs @@ -697,6 +697,73 @@ impl ProgramInfo { } } +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum TransportDirection { + Input, + Output, +} + +fn known_transport_direction(name: &str) -> Option { + match name { + TransportConfig::FILE_INPUT + | TransportConfig::NATS_INPUT + | TransportConfig::KAFKA_INPUT + | TransportConfig::PUB_SUB_INPUT + | TransportConfig::URL_INPUT + | TransportConfig::S3_INPUT + | TransportConfig::DELTA_TABLE_INPUT + | TransportConfig::POSTGRES_INPUT + | TransportConfig::POSTGRES_CDC_INPUT + | TransportConfig::ICEBERG_INPUT + | TransportConfig::DATAGEN + | TransportConfig::NEXMARK + | TransportConfig::HTTP_INPUT + | TransportConfig::ADHOC_INPUT + | TransportConfig::CLOCK + | TransportConfig::EMPTY_INPUT => Some(TransportDirection::Input), + TransportConfig::FILE_OUTPUT + | TransportConfig::KAFKA_OUTPUT + | TransportConfig::DELTA_TABLE_OUTPUT + | TransportConfig::DYNAMODB_OUTPUT + | TransportConfig::POSTGRES_OUTPUT + | TransportConfig::HTTP_OUTPUT + | TransportConfig::REDIS_OUTPUT + | TransportConfig::NULL_OUTPUT => Some(TransportDirection::Output), + _ => None, + } +} + +fn sql_connector_transport_direction(name: &str) -> Option { + match name { + TransportConfig::FILE_INPUT + | TransportConfig::NATS_INPUT + | TransportConfig::KAFKA_INPUT + | TransportConfig::PUB_SUB_INPUT + | TransportConfig::URL_INPUT + | TransportConfig::S3_INPUT + | TransportConfig::DELTA_TABLE_INPUT + | TransportConfig::POSTGRES_INPUT + | TransportConfig::ICEBERG_INPUT + | TransportConfig::DATAGEN + | TransportConfig::NEXMARK + | TransportConfig::EMPTY_INPUT => Some(TransportDirection::Input), + TransportConfig::FILE_OUTPUT + | TransportConfig::KAFKA_OUTPUT + | TransportConfig::DELTA_TABLE_OUTPUT + | TransportConfig::DYNAMODB_OUTPUT + | TransportConfig::POSTGRES_OUTPUT + | TransportConfig::REDIS_OUTPUT + | TransportConfig::NULL_OUTPUT => Some(TransportDirection::Output), + _ => None, + } +} + +fn rejects_sql_connector_transport(name: &str, expected_direction: TransportDirection) -> bool { + let sql_direction = sql_connector_transport_direction(name); + matches!(sql_direction, Some(direction) if direction != expected_direction) + || (known_transport_direction(name).is_some() && sql_direction.is_none()) +} + /// Generates the program info using the program schema. /// The info includes the schema and the input/output connectors derived from it. pub fn generate_program_info( @@ -719,26 +786,13 @@ pub fn generate_program_info( let origin_value = origin_value .clone() .expect("Origin value cannot be None if connectors is non-empty"); - match connector.config.transport { - TransportConfig::FileInput(_) - | TransportConfig::NatsInput(_) - | TransportConfig::KafkaInput(_) - | TransportConfig::PubSubInput(_) - | TransportConfig::UrlInput(_) - | TransportConfig::S3Input(_) - | TransportConfig::DeltaTableInput(_) - | TransportConfig::PostgresInput(_) - | TransportConfig::IcebergInput(_) - | TransportConfig::Datagen(_) - | TransportConfig::Nexmark(_) - | TransportConfig::EmptyInput => {} - _ => { - return Err(ConnectorGenerationError::ExpectedInputConnector { - position: origin_value.value_position, - relation: input_relation.name.sql_name(), - connector_name: connector.name.unwrap_or("".to_string()), - }); - } + let transport_name = connector.config.transport.name(); + if rejects_sql_connector_transport(transport_name, TransportDirection::Input) { + return Err(ConnectorGenerationError::ExpectedInputConnector { + position: origin_value.value_position, + relation: input_relation.name.sql_name(), + connector_name: connector.name.unwrap_or("".to_string()), + }); } input_connectors.push(( input_relation.name.sql_name(), @@ -773,21 +827,13 @@ pub fn generate_program_info( let origin_value = origin_value .clone() .expect("Origin value cannot be None if connectors is non-empty"); - match connector.config.transport { - TransportConfig::FileOutput(_) - | TransportConfig::PostgresOutput(_) - | TransportConfig::KafkaOutput(_) - | TransportConfig::DeltaTableOutput(_) - | TransportConfig::DynamoDBOutput(_) - | TransportConfig::RedisOutput(_) - | TransportConfig::NullOutput => {} - _ => { - return Err(ConnectorGenerationError::ExpectedOutputConnector { - position: origin_value.value_position, - relation: output_relation.name.sql_name(), - connector_name: connector.name.unwrap_or("".to_string()), - }); - } + let transport_name = connector.config.transport.name(); + if rejects_sql_connector_transport(transport_name, TransportDirection::Output) { + return Err(ConnectorGenerationError::ExpectedOutputConnector { + position: origin_value.value_position, + relation: output_relation.name.sql_name(), + connector_name: connector.name.unwrap_or("".to_string()), + }); } output_connectors.push(( output_relation.name.sql_name(), @@ -859,11 +905,17 @@ pub fn generate_pipeline_config( #[cfg(test)] mod tests { - use super::{determine_connector_endpoint_names, RuntimeSelector}; - use crate::db::types::program::ConnectorGenerationError::RelationConnectorNameCollision; + use super::{ + determine_connector_endpoint_names, generate_program_info, known_transport_direction, + sql_connector_transport_direction, RuntimeSelector, + }; + use crate::db::types::program::ConnectorGenerationError::{ + ExpectedInputConnector, ExpectedOutputConnector, RelationConnectorNameCollision, + }; use feldera_types::config::{ConnectorConfig, TransportConfig}; use feldera_types::program_schema::{PropertyValue, SourcePosition}; use feldera_types::transport::datagen::DatagenInputConfig; + use serde_json::json; #[test] fn test_runtime_version_validation() { @@ -895,7 +947,10 @@ mod tests { // Reuse the configuration as it is not used in the function let config = ConnectorConfig { send_snapshot: false, - transport: TransportConfig::Datagen(DatagenInputConfig::default()), + transport: TransportConfig::new( + TransportConfig::DATAGEN, + DatagenInputConfig::default(), + ), format: None, preprocessor: None, postprocessor: None, @@ -986,4 +1041,195 @@ mod tests { } ); } + + fn position() -> SourcePosition { + SourcePosition { + start_line_number: 1, + start_column: 2, + end_line_number: 3, + end_column: 4, + } + } + + fn schema_with_connector(input: bool, transport_name: &str) -> serde_json::Value { + let property_value = PropertyValue { + value: json!([{ + "name": "c1", + "transport": { + "name": transport_name, + } + }]) + .to_string(), + key_position: position(), + value_position: position(), + }; + let relation = json!({ + "name": "t1", + "case_sensitive": false, + "properties": { + "connectors": property_value, + } + }); + + if input { + json!({ + "inputs": [relation], + "outputs": [], + }) + } else { + json!({ + "inputs": [], + "outputs": [relation], + }) + } + } + + fn builtin_transport_names() -> [&'static str; 24] { + [ + TransportConfig::FILE_INPUT, + TransportConfig::FILE_OUTPUT, + TransportConfig::NATS_INPUT, + TransportConfig::KAFKA_INPUT, + TransportConfig::KAFKA_OUTPUT, + TransportConfig::PUB_SUB_INPUT, + TransportConfig::URL_INPUT, + TransportConfig::S3_INPUT, + TransportConfig::DELTA_TABLE_INPUT, + TransportConfig::DELTA_TABLE_OUTPUT, + TransportConfig::DYNAMODB_OUTPUT, + TransportConfig::REDIS_OUTPUT, + TransportConfig::ICEBERG_INPUT, + TransportConfig::POSTGRES_INPUT, + TransportConfig::POSTGRES_CDC_INPUT, + TransportConfig::POSTGRES_OUTPUT, + TransportConfig::DATAGEN, + TransportConfig::NEXMARK, + TransportConfig::HTTP_INPUT, + TransportConfig::HTTP_OUTPUT, + TransportConfig::ADHOC_INPUT, + TransportConfig::CLOCK, + TransportConfig::NULL_OUTPUT, + TransportConfig::EMPTY_INPUT, + ] + } + + #[test] + fn builtin_transports_have_known_direction() { + for transport_name in builtin_transport_names() { + assert!( + known_transport_direction(transport_name).is_some(), + "{transport_name} should have a known transport direction" + ); + } + } + + #[test] + fn sql_transport_directions_match_known_directions() { + for transport_name in builtin_transport_names() { + let Some(sql_direction) = sql_connector_transport_direction(transport_name) else { + continue; + }; + assert_eq!( + known_transport_direction(transport_name), + Some(sql_direction), + "{transport_name} should have matching SQL and known transport directions" + ); + } + } + + #[test] + fn generate_program_info_rejects_known_output_transport_on_input_relation() { + assert_eq!( + generate_program_info( + schema_with_connector(true, TransportConfig::NULL_OUTPUT), + String::new(), + String::new(), + None, + ) + .unwrap_err(), + ExpectedInputConnector { + position: position(), + relation: "t1".to_string(), + connector_name: "c1".to_string(), + } + ); + } + + #[test] + fn generate_program_info_rejects_runtime_only_input_transports() { + for transport_name in [ + TransportConfig::HTTP_INPUT, + TransportConfig::ADHOC_INPUT, + TransportConfig::CLOCK, + ] { + assert_eq!( + generate_program_info( + schema_with_connector(true, transport_name), + String::new(), + String::new(), + None, + ) + .unwrap_err(), + ExpectedInputConnector { + position: position(), + relation: "t1".to_string(), + connector_name: "c1".to_string(), + } + ); + } + } + + #[test] + fn generate_program_info_rejects_known_input_transport_on_output_relation() { + assert_eq!( + generate_program_info( + schema_with_connector(false, TransportConfig::DATAGEN), + String::new(), + String::new(), + None, + ) + .unwrap_err(), + ExpectedOutputConnector { + position: position(), + relation: "t1".to_string(), + connector_name: "c1".to_string(), + } + ); + } + + #[test] + fn generate_program_info_rejects_runtime_only_output_transports() { + assert_eq!( + generate_program_info( + schema_with_connector(false, TransportConfig::HTTP_OUTPUT), + String::new(), + String::new(), + None, + ) + .unwrap_err(), + ExpectedOutputConnector { + position: position(), + relation: "t1".to_string(), + connector_name: "c1".to_string(), + } + ); + } + + #[test] + fn generate_program_info_allows_unknown_dynamic_transport_names() { + assert!(generate_program_info( + schema_with_connector(true, "custom_input"), + String::new(), + String::new(), + None, + ) + .is_ok()); + assert!(generate_program_info( + schema_with_connector(false, "custom_output"), + String::new(), + String::new(), + None, + ) + .is_ok()); + } }