From 54a7c82f7aae17984023c30eaf1231f7238cba52 Mon Sep 17 00:00:00 2001 From: Dnreikronos Date: Mon, 22 Jun 2026 17:33:10 -0300 Subject: [PATCH 1/6] Add connector transport registries --- crates/adapterlib/src/catalog.rs | 7 + crates/adapterlib/src/transport.rs | 96 +++++- crates/adapters/src/catalog.rs | 20 +- crates/adapters/src/controller.rs | 44 ++- crates/adapters/src/integrated.rs | 480 +++++++++++++++++++++++++---- crates/adapters/src/transport.rs | 454 +++++++++++++++++++++++---- 6 files changed, 969 insertions(+), 132 deletions(-) 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..7225408b055 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<&'static str, Arc>, +} + +impl InputTransportRegistry { + pub fn new() -> Self { + Self { + registered: BTreeMap::new(), + } + } + + pub fn register( + &mut self, + name: &'static str, + factory: Box, + ) { + self.registered.insert(name, 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<&'static str, Arc>, +} + +impl OutputTransportRegistry { + pub fn new() -> Self { + Self { + registered: BTreeMap::new(), + } + } + + pub fn register( + &mut self, + name: &'static str, + factory: Box, + ) { + self.registered.insert(name, 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/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..8864a11540f 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}; @@ -6268,12 +6269,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`. @@ -6588,13 +6600,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/integrated.rs b/crates/adapters/src/integrated.rs index a36df0f8d3d..baea053d1d3 100644 --- a/crates/adapters/src/integrated.rs +++ b/crates/adapters/src/integrated.rs @@ -4,7 +4,10 @@ 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 std::sync::{Arc, Weak}; +use std::{ + collections::BTreeMap, + sync::{Arc, Weak}, +}; #[cfg(feature = "with-deltalake")] pub mod delta_table; @@ -40,6 +43,354 @@ where } } +/// Factory for creating integrated output endpoints from connector configuration. +pub trait IntegratedOutputEndpointFactory: Send + Sync { + #[allow(clippy::too_many_arguments)] + fn create( + &self, + endpoint_id: EndpointId, + endpoint_name: &str, + connector_config: &ConnectorConfig, + key_schema: &Option, + schema: &Relation, + controller: Weak, + continue_previous_state: bool, + is_index: bool, + ) -> Result>, ControllerError>; +} + +/// Registry of integrated output endpoint factories keyed by transport name. +#[derive(Default)] +pub struct IntegratedOutputEndpointRegistry { + registered: BTreeMap<&'static str, Arc>, +} + +impl IntegratedOutputEndpointRegistry { + pub fn new() -> Self { + Self { + registered: BTreeMap::new(), + } + } + + pub fn register( + &mut self, + name: &'static str, + factory: Box, + ) { + self.registered.insert(name, Arc::from(factory)); + } + + pub fn get(&self, name: &str) -> Option> { + self.registered.get(name).cloned() + } + + #[allow(clippy::too_many_arguments)] + pub fn create_endpoint( + &self, + endpoint_id: EndpointId, + endpoint_name: &str, + connector_config: &ConnectorConfig, + key_schema: &Option, + schema: &Relation, + controller: Weak, + continue_previous_state: bool, + is_index: bool, + ) -> Result>, ControllerError> { + let Some(factory) = self.get(&connector_config.transport.name()) else { + return Ok(None); + }; + factory.create( + endpoint_id, + endpoint_name, + connector_config, + key_schema, + schema, + controller, + continue_previous_state, + is_index, + ) + } +} + +/// Factory for creating integrated input endpoints from connector configuration. +pub trait IntegratedInputEndpointFactory: Send + Sync { + fn create( + &self, + endpoint_name: &str, + config: &ConnectorConfig, + pipeline_config: &PipelineConfig, + runtime_env: Arc, + consumer: Box, + ) -> Result>, ControllerError>; +} + +/// Registry of integrated input endpoint factories keyed by transport name. +#[derive(Default)] +pub struct IntegratedInputEndpointRegistry { + registered: BTreeMap<&'static str, Arc>, +} + +impl IntegratedInputEndpointRegistry { + pub fn new() -> Self { + Self { + registered: BTreeMap::new(), + } + } + + pub fn register( + &mut self, + name: &'static str, + factory: Box, + ) { + self.registered.insert(name, Arc::from(factory)); + } + + pub fn get(&self, name: &str) -> Option> { + self.registered.get(name).cloned() + } + + pub fn create_endpoint( + &self, + endpoint_name: &str, + config: &ConnectorConfig, + pipeline_config: &PipelineConfig, + runtime_env: Arc, + consumer: Box, + ) -> Result>, ControllerError> { + let Some(factory) = self.get(&config.transport.name()) else { + return Ok(None); + }; + factory.create( + endpoint_name, + config, + pipeline_config, + runtime_env, + consumer, + ) + } +} + +pub fn builtin_integrated_output_endpoint_registry() -> IntegratedOutputEndpointRegistry { + let mut registry = IntegratedOutputEndpointRegistry::new(); + #[cfg(feature = "with-deltalake")] + registry.register("delta_table_output", Box::new(DeltaTableOutputFactory)); + registry.register("postgres_output", Box::new(PostgresOutputFactory)); + #[cfg(feature = "with-dynamodb")] + registry.register("dynamodb_output", Box::new(DynamoDBOutputFactory)); + registry +} + +pub fn builtin_integrated_input_endpoint_registry() -> IntegratedInputEndpointRegistry { + let mut registry = IntegratedInputEndpointRegistry::new(); + #[cfg(feature = "with-deltalake")] + registry.register("delta_table_input", Box::new(DeltaTableInputFactory)); + #[cfg(feature = "with-iceberg")] + registry.register("iceberg_input", Box::new(IcebergInputFactory)); + registry.register("postgres_input", Box::new(PostgresInputFactory)); + #[cfg(feature = "with-postgres-cdc")] + registry.register("postgres_cdc_input", Box::new(PostgresCdcInputFactory)); + registry +} + +#[cfg(feature = "with-deltalake")] +struct DeltaTableOutputFactory; + +#[cfg(feature = "with-deltalake")] +impl IntegratedOutputEndpointFactory for DeltaTableOutputFactory { + #[allow(clippy::too_many_arguments)] + fn create( + &self, + endpoint_id: EndpointId, + endpoint_name: &str, + connector_config: &ConnectorConfig, + key_schema: &Option, + schema: &Relation, + controller: Weak, + continue_previous_state: bool, + is_index: bool, + ) -> Result>, ControllerError> { + match &connector_config.transport { + TransportConfig::DeltaTableOutput(config) => { + Ok(Some(Box::new(delta_table::DeltaTableWriter::new( + endpoint_id, + endpoint_name, + config, + key_schema, + schema, + controller, + continue_previous_state, + is_index, + )?))) + } + _ => Ok(None), + } + } +} + +struct PostgresOutputFactory; + +impl IntegratedOutputEndpointFactory for PostgresOutputFactory { + #[allow(clippy::too_many_arguments)] + fn create( + &self, + endpoint_id: EndpointId, + endpoint_name: &str, + connector_config: &ConnectorConfig, + key_schema: &Option, + schema: &Relation, + controller: Weak, + _continue_previous_state: bool, + is_index: bool, + ) -> Result>, ControllerError> { + match &connector_config.transport { + TransportConfig::PostgresOutput(config) => { + Ok(Some(Box::new(PostgresOutputEndpoint::new( + endpoint_id, + endpoint_name, + config, + key_schema, + schema, + controller, + is_index, + )?))) + } + _ => Ok(None), + } + } +} + +#[cfg(feature = "with-dynamodb")] +struct DynamoDBOutputFactory; + +#[cfg(feature = "with-dynamodb")] +impl IntegratedOutputEndpointFactory for DynamoDBOutputFactory { + #[allow(clippy::too_many_arguments)] + fn create( + &self, + endpoint_id: EndpointId, + endpoint_name: &str, + connector_config: &ConnectorConfig, + key_schema: &Option, + schema: &Relation, + controller: Weak, + _continue_previous_state: bool, + is_index: bool, + ) -> Result>, ControllerError> { + match &connector_config.transport { + TransportConfig::DynamoDBOutput(config) => { + Ok(Some(Box::new(DynamoDBOutputEndpoint::new( + endpoint_id, + endpoint_name, + config, + key_schema, + schema, + controller, + is_index, + )?))) + } + _ => Ok(None), + } + } +} + +#[cfg(feature = "with-deltalake")] +struct DeltaTableInputFactory; + +#[cfg(feature = "with-deltalake")] +impl IntegratedInputEndpointFactory for DeltaTableInputFactory { + fn create( + &self, + endpoint_name: &str, + config: &ConnectorConfig, + pipeline_config: &PipelineConfig, + runtime_env: Arc, + consumer: Box, + ) -> Result>, ControllerError> { + match &config.transport { + TransportConfig::DeltaTableInput(config) => { + Ok(Some(Box::new(delta_table::DeltaTableInputEndpoint::new( + endpoint_name, + config, + pipeline_config, + runtime_env, + consumer, + )))) + } + _ => Ok(None), + } + } +} + +#[cfg(feature = "with-iceberg")] +struct IcebergInputFactory; + +#[cfg(feature = "with-iceberg")] +impl IntegratedInputEndpointFactory for IcebergInputFactory { + fn create( + &self, + endpoint_name: &str, + config: &ConnectorConfig, + pipeline_config: &PipelineConfig, + runtime_env: Arc, + consumer: Box, + ) -> Result>, ControllerError> { + match &config.transport { + TransportConfig::IcebergInput(config) => { + Ok(Some(Box::new(feldera_iceberg::IcebergInputEndpoint::new( + endpoint_name, + config, + pipeline_config, + runtime_env, + consumer, + )))) + } + _ => Ok(None), + } + } +} + +struct PostgresInputFactory; + +impl IntegratedInputEndpointFactory for PostgresInputFactory { + fn create( + &self, + endpoint_name: &str, + config: &ConnectorConfig, + _pipeline_config: &PipelineConfig, + _runtime_env: Arc, + consumer: Box, + ) -> Result>, ControllerError> { + match &config.transport { + TransportConfig::PostgresInput(config) => Ok(Some(Box::new( + PostgresInputEndpoint::new(endpoint_name, config, consumer), + ))), + _ => Ok(None), + } + } +} + +#[cfg(feature = "with-postgres-cdc")] +struct PostgresCdcInputFactory; + +#[cfg(feature = "with-postgres-cdc")] +impl IntegratedInputEndpointFactory for PostgresCdcInputFactory { + fn create( + &self, + endpoint_name: &str, + config: &ConnectorConfig, + _pipeline_config: &PipelineConfig, + _runtime_env: Arc, + consumer: Box, + ) -> Result>, ControllerError> { + match &config.transport { + TransportConfig::PostgresCdcInput(config) => Ok(Some(Box::new( + PostgresCdcInputEndpoint::new(endpoint_name, config, consumer), + ))), + _ => Ok(None), + } + } +} + /// Create an instance of an integrated output endpoint given its config /// and output relation schema. #[allow(unused, clippy::too_many_arguments)] @@ -53,44 +404,49 @@ pub fn create_integrated_output_endpoint( continue_previous_state: bool, is_index: bool, ) -> Result, ControllerError> { - let ep: Box = match &connector_config.transport { - #[cfg(feature = "with-deltalake")] - TransportConfig::DeltaTableOutput(config) => Box::new(delta_table::DeltaTableWriter::new( + let registry = builtin_integrated_output_endpoint_registry(); + create_integrated_output_endpoint_with_registry( + ®istry, + endpoint_id, + endpoint_name, + connector_config, + key_schema, + schema, + controller, + continue_previous_state, + is_index, + ) +} + +#[allow(unused, clippy::too_many_arguments)] +pub fn create_integrated_output_endpoint_with_registry( + registry: &IntegratedOutputEndpointRegistry, + endpoint_id: EndpointId, + endpoint_name: &str, + connector_config: &ConnectorConfig, + key_schema: &Option, + schema: &Relation, + controller: Weak, + continue_previous_state: bool, + is_index: bool, +) -> Result, ControllerError> { + let ep = registry + .create_endpoint( endpoint_id, endpoint_name, - config, + connector_config, key_schema, schema, controller, continue_previous_state, is_index, - )?), - TransportConfig::PostgresOutput(config) => Box::new(PostgresOutputEndpoint::new( - endpoint_id, - endpoint_name, - config, - key_schema, - schema, - controller, - is_index, - )?), - #[cfg(feature = "with-dynamodb")] - TransportConfig::DynamoDBOutput(config) => Box::new(DynamoDBOutputEndpoint::new( - endpoint_id, - endpoint_name, - config, - key_schema, - schema, - controller, - is_index, - )?), - transport => { - return Err(ControllerError::unknown_output_transport( + )? + .ok_or_else(|| { + ControllerError::unknown_output_transport( endpoint_name, - &transport.name(), - )); - } - }; + &connector_config.transport.name(), + ) + })?; if connector_config.format.is_some() { return Err(ControllerError::invalid_parser_configuration( @@ -113,43 +469,37 @@ pub fn create_integrated_input_endpoint( runtime_env: Arc, consumer: Box, ) -> Result, ControllerError> { - let ep: Box = match &config.transport { - #[cfg(feature = "with-deltalake")] - TransportConfig::DeltaTableInput(config) => { - Box::new(delta_table::DeltaTableInputEndpoint::new( - endpoint_name, - config, - 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)) - } - #[cfg(feature = "with-postgres-cdc")] - TransportConfig::PostgresCdcInput(config) => Box::new(PostgresCdcInputEndpoint::new( + let registry = builtin_integrated_input_endpoint_registry(); + create_integrated_input_endpoint_with_registry( + ®istry, + endpoint_name, + config, + pipeline_config, + runtime_env, + consumer, + ) +} + +#[allow(unused_variables)] +pub fn create_integrated_input_endpoint_with_registry( + registry: &IntegratedInputEndpointRegistry, + endpoint_name: &str, + config: &ConnectorConfig, + pipeline_config: &PipelineConfig, + runtime_env: Arc, + consumer: Box, +) -> Result, ControllerError> { + let ep = registry + .create_endpoint( endpoint_name, config, + pipeline_config, + runtime_env, consumer, - )), - transport => { - return Err(ControllerError::unknown_input_transport( - endpoint_name, - &transport.name(), - )); - } - }; + )? + .ok_or_else(|| { + ControllerError::unknown_input_transport(endpoint_name, &config.transport.name()) + })?; if config.format.is_some() { return Err(ControllerError::invalid_parser_configuration( diff --git a/crates/adapters/src/transport.rs b/crates/adapters/src/transport.rs index f634e86c0e4..d5ad42d7b21 100644 --- a/crates/adapters/src/transport.rs +++ b/crates/adapters/src/transport.rs @@ -80,6 +80,308 @@ use feldera_datagen::GeneratorEndpoint; pub use feldera_adapterlib::transport::*; +pub fn builtin_input_transport_registry() -> InputTransportRegistry { + let mut registry = InputTransportRegistry::new(); + registry.register("file_input", Box::new(FileInputFactory)); + #[cfg(feature = "with-kafka")] + registry.register("kafka_input", Box::new(KafkaInputFactory)); + #[cfg(feature = "with-nats")] + registry.register("nats_input", Box::new(NatsInputFactory)); + #[cfg(feature = "with-pubsub")] + registry.register("pub_sub_input", Box::new(PubSubInputFactory)); + registry.register("url_input", Box::new(UrlInputFactory)); + registry.register("s3_input", Box::new(S3InputFactory)); + registry.register("datagen", Box::new(DatagenInputFactory)); + #[cfg(feature = "with-nexmark")] + registry.register("nexmark", Box::new(NexmarkInputFactory)); + registry.register("http_input", Box::new(HttpInputFactory)); + registry.register("adhoc_input", Box::new(AdHocInputFactory)); + registry.register("clock", Box::new(ClockInputFactory)); + registry.register("empty_input", Box::new(EmptyInputFactory)); + registry +} + +pub fn builtin_output_transport_registry() -> OutputTransportRegistry { + let mut registry = OutputTransportRegistry::new(); + registry.register("file_output", Box::new(FileOutputFactory)); + #[cfg(feature = "with-kafka")] + registry.register("kafka_output", Box::new(KafkaOutputFactory)); + #[cfg(feature = "with-redis")] + registry.register("redis_output", Box::new(RedisOutputFactory)); + registry.register("null_output", Box::new(NullOutputFactory)); + registry +} + +struct FileInputFactory; + +impl InputTransportEndpointFactory for FileInputFactory { + fn create( + &self, + config: &TransportConfig, + ) -> AnyResult>> { + match config { + TransportConfig::FileInput(config) => { + Ok(Some(Box::new(FileInputEndpoint::new(config)))) + } + _ => Ok(None), + } + } +} + +#[cfg(feature = "with-kafka")] +struct KafkaInputFactory; + +#[cfg(feature = "with-kafka")] +impl InputTransportEndpointFactory for KafkaInputFactory { + fn create( + &self, + config: &TransportConfig, + ) -> AnyResult>> { + match config { + TransportConfig::KafkaInput(config) => { + Ok(Some(Box::new(KafkaFtInputEndpoint::new(config)?))) + } + _ => Ok(None), + } + } +} + +#[cfg(feature = "with-nats")] +struct NatsInputFactory; + +#[cfg(feature = "with-nats")] +impl InputTransportEndpointFactory for NatsInputFactory { + fn create( + &self, + config: &TransportConfig, + ) -> AnyResult>> { + match config { + TransportConfig::NatsInput(config) => { + Ok(Some(Box::new(NatsInputEndpoint::new(config)?))) + } + _ => Ok(None), + } + } +} + +#[cfg(feature = "with-pubsub")] +struct PubSubInputFactory; + +#[cfg(feature = "with-pubsub")] +impl InputTransportEndpointFactory for PubSubInputFactory { + fn create( + &self, + config: &TransportConfig, + ) -> AnyResult>> { + match config { + TransportConfig::PubSubInput(config) => { + Ok(Some(Box::new(PubSubInputEndpoint::new(config.clone())?))) + } + _ => Ok(None), + } + } +} + +struct UrlInputFactory; + +impl InputTransportEndpointFactory for UrlInputFactory { + fn create( + &self, + config: &TransportConfig, + ) -> AnyResult>> { + match config { + TransportConfig::UrlInput(config) => Ok(Some(Box::new(UrlInputEndpoint::new(config)))), + _ => Ok(None), + } + } +} + +struct S3InputFactory; + +impl InputTransportEndpointFactory for S3InputFactory { + fn create( + &self, + config: &TransportConfig, + ) -> AnyResult>> { + match config { + TransportConfig::S3Input(config) => Ok(Some(Box::new(S3InputEndpoint::new(config)?))), + _ => Ok(None), + } + } +} + +struct DatagenInputFactory; + +impl InputTransportEndpointFactory for DatagenInputFactory { + fn create( + &self, + config: &TransportConfig, + ) -> AnyResult>> { + match config { + TransportConfig::Datagen(config) => { + Ok(Some(Box::new(GeneratorEndpoint::new(config.clone())))) + } + _ => Ok(None), + } + } +} + +#[cfg(feature = "with-nexmark")] +struct NexmarkInputFactory; + +#[cfg(feature = "with-nexmark")] +impl InputTransportEndpointFactory for NexmarkInputFactory { + fn create( + &self, + config: &TransportConfig, + ) -> AnyResult>> { + match config { + TransportConfig::Nexmark(config) => { + Ok(Some(Box::new(NexmarkEndpoint::new(config.clone())))) + } + _ => Ok(None), + } + } +} + +struct HttpInputFactory; + +impl InputTransportEndpointFactory for HttpInputFactory { + fn create( + &self, + config: &TransportConfig, + ) -> AnyResult>> { + match config { + TransportConfig::HttpInput(config) => { + Ok(Some(Box::new(HttpInputEndpoint::new(config)))) + } + _ => Ok(None), + } + } +} + +struct AdHocInputFactory; + +impl InputTransportEndpointFactory for AdHocInputFactory { + fn create( + &self, + config: &TransportConfig, + ) -> AnyResult>> { + match config { + TransportConfig::AdHocInput(config) => { + Ok(Some(Box::new(AdHocInputEndpoint::new(config)))) + } + _ => Ok(None), + } + } +} + +struct ClockInputFactory; + +impl InputTransportEndpointFactory for ClockInputFactory { + fn create( + &self, + config: &TransportConfig, + ) -> AnyResult>> { + match config { + TransportConfig::ClockInput(config) => Ok(Some(Box::new(ClockEndpoint::new(config)?))), + _ => Ok(None), + } + } +} + +struct EmptyInputFactory; + +impl InputTransportEndpointFactory for EmptyInputFactory { + fn create( + &self, + config: &TransportConfig, + ) -> AnyResult>> { + match config { + TransportConfig::EmptyInput => Ok(Some(Box::new(EmptyInputEndpoint))), + _ => Ok(None), + } + } +} + +struct FileOutputFactory; + +impl OutputTransportEndpointFactory for FileOutputFactory { + fn create( + &self, + config: &TransportConfig, + _endpoint_name: &str, + _fault_tolerant: bool, + ) -> AnyResult>> { + match config { + TransportConfig::FileOutput(config) => { + Ok(Some(Box::new(FileOutputEndpoint::new(config)?))) + } + _ => Ok(None), + } + } +} + +#[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>> { + match config { + TransportConfig::KafkaOutput(config) => match fault_tolerant { + false => Ok(Some(Box::new(KafkaOutputEndpoint::new( + config, + endpoint_name, + )?))), + true => Ok(Some(Box::new(KafkaFtOutputEndpoint::new(config)?))), + }, + _ => Ok(None), + } + } +} + +#[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>> { + match config { + TransportConfig::RedisOutput(config) => { + Ok(Some(Box::new(RedisOutputEndpoint::new(config)?))) + } + _ => Ok(None), + } + } +} + +struct NullOutputFactory; + +impl OutputTransportEndpointFactory for NullOutputFactory { + fn create( + &self, + config: &TransportConfig, + _endpoint_name: &str, + _fault_tolerant: bool, + ) -> AnyResult>> { + match config { + TransportConfig::NullOutput => Ok(Some(Box::new(NullOutputEndpoint))), + _ => Ok(None), + } + } +} + /// Creates an input transport endpoint instance using an input transport /// configuration, resolving secrets by reading `secrets_dir`. /// @@ -92,45 +394,7 @@ pub fn input_transport_config_to_endpoint( secrets_dir: &Path, ) -> 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)) + builtin_input_transport_registry().create_endpoint(&config) } /// Creates an output transport endpoint instance using an output transport @@ -150,21 +414,101 @@ pub fn output_transport_config_to_endpoint( secrets_dir: &Path, ) -> AnyResult>> { let config = resolve_secret_references_via_json(secrets_dir, config)?; - match config { - TransportConfig::FileOutput(config) => Ok(Some(Box::new(FileOutputEndpoint::new(config)?))), - #[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)?))), - }, - #[cfg(feature = "with-redis")] - TransportConfig::RedisOutput(config) => { - Ok(Some(Box::new(RedisOutputEndpoint::new(config)?))) - } - TransportConfig::NullOutput => Ok(Some(Box::new(NullOutputEndpoint))), - _ => Ok(None), + builtin_output_transport_registry().create_endpoint(&config, endpoint_name, fault_tolerant) +} + +#[cfg(test)] +mod tests { + use super::*; + use feldera_types::config::FtModel; + + #[test] + fn builtin_input_registry_creates_empty_input_endpoint() { + let secrets_dir = tempfile::tempdir().unwrap(); + + let endpoint = input_transport_config_to_endpoint( + &TransportConfig::EmptyInput, + "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::NullOutput, + "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::NullOutput, + "null", + secrets_dir.path() + ) + .unwrap() + .is_none() + ); + assert!( + output_transport_config_to_endpoint( + &TransportConfig::EmptyInput, + "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::EmptyInput) + .unwrap() + .is_none() + ); + input_registry.register("empty_input", Box::new(EmptyInputFactory)); + assert!( + input_registry + .create_endpoint(&TransportConfig::EmptyInput) + .unwrap() + .is_some() + ); + + let mut output_registry = OutputTransportRegistry::new(); + assert!( + output_registry + .create_endpoint(&TransportConfig::NullOutput, "null", true) + .unwrap() + .is_none() + ); + output_registry.register("null_output", Box::new(NullOutputFactory)); + assert!( + output_registry + .create_endpoint(&TransportConfig::NullOutput, "null", true) + .unwrap() + .is_some() + ); } } From c71dabfef0a6df163324179f2f612f86fc90caea Mon Sep 17 00:00:00 2001 From: Dnreikronos Date: Mon, 22 Jun 2026 17:57:25 -0300 Subject: [PATCH 2/6] Tighten transport registry scope --- crates/adapters/src/integrated.rs | 480 ++++------------------------- crates/adapters/src/transport.rs | 65 ++-- crates/feldera-types/src/config.rs | 73 +++-- 3 files changed, 161 insertions(+), 457 deletions(-) diff --git a/crates/adapters/src/integrated.rs b/crates/adapters/src/integrated.rs index baea053d1d3..a36df0f8d3d 100644 --- a/crates/adapters/src/integrated.rs +++ b/crates/adapters/src/integrated.rs @@ -4,10 +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 std::{ - collections::BTreeMap, - sync::{Arc, Weak}, -}; +use std::sync::{Arc, Weak}; #[cfg(feature = "with-deltalake")] pub mod delta_table; @@ -43,354 +40,6 @@ where } } -/// Factory for creating integrated output endpoints from connector configuration. -pub trait IntegratedOutputEndpointFactory: Send + Sync { - #[allow(clippy::too_many_arguments)] - fn create( - &self, - endpoint_id: EndpointId, - endpoint_name: &str, - connector_config: &ConnectorConfig, - key_schema: &Option, - schema: &Relation, - controller: Weak, - continue_previous_state: bool, - is_index: bool, - ) -> Result>, ControllerError>; -} - -/// Registry of integrated output endpoint factories keyed by transport name. -#[derive(Default)] -pub struct IntegratedOutputEndpointRegistry { - registered: BTreeMap<&'static str, Arc>, -} - -impl IntegratedOutputEndpointRegistry { - pub fn new() -> Self { - Self { - registered: BTreeMap::new(), - } - } - - pub fn register( - &mut self, - name: &'static str, - factory: Box, - ) { - self.registered.insert(name, Arc::from(factory)); - } - - pub fn get(&self, name: &str) -> Option> { - self.registered.get(name).cloned() - } - - #[allow(clippy::too_many_arguments)] - pub fn create_endpoint( - &self, - endpoint_id: EndpointId, - endpoint_name: &str, - connector_config: &ConnectorConfig, - key_schema: &Option, - schema: &Relation, - controller: Weak, - continue_previous_state: bool, - is_index: bool, - ) -> Result>, ControllerError> { - let Some(factory) = self.get(&connector_config.transport.name()) else { - return Ok(None); - }; - factory.create( - endpoint_id, - endpoint_name, - connector_config, - key_schema, - schema, - controller, - continue_previous_state, - is_index, - ) - } -} - -/// Factory for creating integrated input endpoints from connector configuration. -pub trait IntegratedInputEndpointFactory: Send + Sync { - fn create( - &self, - endpoint_name: &str, - config: &ConnectorConfig, - pipeline_config: &PipelineConfig, - runtime_env: Arc, - consumer: Box, - ) -> Result>, ControllerError>; -} - -/// Registry of integrated input endpoint factories keyed by transport name. -#[derive(Default)] -pub struct IntegratedInputEndpointRegistry { - registered: BTreeMap<&'static str, Arc>, -} - -impl IntegratedInputEndpointRegistry { - pub fn new() -> Self { - Self { - registered: BTreeMap::new(), - } - } - - pub fn register( - &mut self, - name: &'static str, - factory: Box, - ) { - self.registered.insert(name, Arc::from(factory)); - } - - pub fn get(&self, name: &str) -> Option> { - self.registered.get(name).cloned() - } - - pub fn create_endpoint( - &self, - endpoint_name: &str, - config: &ConnectorConfig, - pipeline_config: &PipelineConfig, - runtime_env: Arc, - consumer: Box, - ) -> Result>, ControllerError> { - let Some(factory) = self.get(&config.transport.name()) else { - return Ok(None); - }; - factory.create( - endpoint_name, - config, - pipeline_config, - runtime_env, - consumer, - ) - } -} - -pub fn builtin_integrated_output_endpoint_registry() -> IntegratedOutputEndpointRegistry { - let mut registry = IntegratedOutputEndpointRegistry::new(); - #[cfg(feature = "with-deltalake")] - registry.register("delta_table_output", Box::new(DeltaTableOutputFactory)); - registry.register("postgres_output", Box::new(PostgresOutputFactory)); - #[cfg(feature = "with-dynamodb")] - registry.register("dynamodb_output", Box::new(DynamoDBOutputFactory)); - registry -} - -pub fn builtin_integrated_input_endpoint_registry() -> IntegratedInputEndpointRegistry { - let mut registry = IntegratedInputEndpointRegistry::new(); - #[cfg(feature = "with-deltalake")] - registry.register("delta_table_input", Box::new(DeltaTableInputFactory)); - #[cfg(feature = "with-iceberg")] - registry.register("iceberg_input", Box::new(IcebergInputFactory)); - registry.register("postgres_input", Box::new(PostgresInputFactory)); - #[cfg(feature = "with-postgres-cdc")] - registry.register("postgres_cdc_input", Box::new(PostgresCdcInputFactory)); - registry -} - -#[cfg(feature = "with-deltalake")] -struct DeltaTableOutputFactory; - -#[cfg(feature = "with-deltalake")] -impl IntegratedOutputEndpointFactory for DeltaTableOutputFactory { - #[allow(clippy::too_many_arguments)] - fn create( - &self, - endpoint_id: EndpointId, - endpoint_name: &str, - connector_config: &ConnectorConfig, - key_schema: &Option, - schema: &Relation, - controller: Weak, - continue_previous_state: bool, - is_index: bool, - ) -> Result>, ControllerError> { - match &connector_config.transport { - TransportConfig::DeltaTableOutput(config) => { - Ok(Some(Box::new(delta_table::DeltaTableWriter::new( - endpoint_id, - endpoint_name, - config, - key_schema, - schema, - controller, - continue_previous_state, - is_index, - )?))) - } - _ => Ok(None), - } - } -} - -struct PostgresOutputFactory; - -impl IntegratedOutputEndpointFactory for PostgresOutputFactory { - #[allow(clippy::too_many_arguments)] - fn create( - &self, - endpoint_id: EndpointId, - endpoint_name: &str, - connector_config: &ConnectorConfig, - key_schema: &Option, - schema: &Relation, - controller: Weak, - _continue_previous_state: bool, - is_index: bool, - ) -> Result>, ControllerError> { - match &connector_config.transport { - TransportConfig::PostgresOutput(config) => { - Ok(Some(Box::new(PostgresOutputEndpoint::new( - endpoint_id, - endpoint_name, - config, - key_schema, - schema, - controller, - is_index, - )?))) - } - _ => Ok(None), - } - } -} - -#[cfg(feature = "with-dynamodb")] -struct DynamoDBOutputFactory; - -#[cfg(feature = "with-dynamodb")] -impl IntegratedOutputEndpointFactory for DynamoDBOutputFactory { - #[allow(clippy::too_many_arguments)] - fn create( - &self, - endpoint_id: EndpointId, - endpoint_name: &str, - connector_config: &ConnectorConfig, - key_schema: &Option, - schema: &Relation, - controller: Weak, - _continue_previous_state: bool, - is_index: bool, - ) -> Result>, ControllerError> { - match &connector_config.transport { - TransportConfig::DynamoDBOutput(config) => { - Ok(Some(Box::new(DynamoDBOutputEndpoint::new( - endpoint_id, - endpoint_name, - config, - key_schema, - schema, - controller, - is_index, - )?))) - } - _ => Ok(None), - } - } -} - -#[cfg(feature = "with-deltalake")] -struct DeltaTableInputFactory; - -#[cfg(feature = "with-deltalake")] -impl IntegratedInputEndpointFactory for DeltaTableInputFactory { - fn create( - &self, - endpoint_name: &str, - config: &ConnectorConfig, - pipeline_config: &PipelineConfig, - runtime_env: Arc, - consumer: Box, - ) -> Result>, ControllerError> { - match &config.transport { - TransportConfig::DeltaTableInput(config) => { - Ok(Some(Box::new(delta_table::DeltaTableInputEndpoint::new( - endpoint_name, - config, - pipeline_config, - runtime_env, - consumer, - )))) - } - _ => Ok(None), - } - } -} - -#[cfg(feature = "with-iceberg")] -struct IcebergInputFactory; - -#[cfg(feature = "with-iceberg")] -impl IntegratedInputEndpointFactory for IcebergInputFactory { - fn create( - &self, - endpoint_name: &str, - config: &ConnectorConfig, - pipeline_config: &PipelineConfig, - runtime_env: Arc, - consumer: Box, - ) -> Result>, ControllerError> { - match &config.transport { - TransportConfig::IcebergInput(config) => { - Ok(Some(Box::new(feldera_iceberg::IcebergInputEndpoint::new( - endpoint_name, - config, - pipeline_config, - runtime_env, - consumer, - )))) - } - _ => Ok(None), - } - } -} - -struct PostgresInputFactory; - -impl IntegratedInputEndpointFactory for PostgresInputFactory { - fn create( - &self, - endpoint_name: &str, - config: &ConnectorConfig, - _pipeline_config: &PipelineConfig, - _runtime_env: Arc, - consumer: Box, - ) -> Result>, ControllerError> { - match &config.transport { - TransportConfig::PostgresInput(config) => Ok(Some(Box::new( - PostgresInputEndpoint::new(endpoint_name, config, consumer), - ))), - _ => Ok(None), - } - } -} - -#[cfg(feature = "with-postgres-cdc")] -struct PostgresCdcInputFactory; - -#[cfg(feature = "with-postgres-cdc")] -impl IntegratedInputEndpointFactory for PostgresCdcInputFactory { - fn create( - &self, - endpoint_name: &str, - config: &ConnectorConfig, - _pipeline_config: &PipelineConfig, - _runtime_env: Arc, - consumer: Box, - ) -> Result>, ControllerError> { - match &config.transport { - TransportConfig::PostgresCdcInput(config) => Ok(Some(Box::new( - PostgresCdcInputEndpoint::new(endpoint_name, config, consumer), - ))), - _ => Ok(None), - } - } -} - /// Create an instance of an integrated output endpoint given its config /// and output relation schema. #[allow(unused, clippy::too_many_arguments)] @@ -404,49 +53,44 @@ pub fn create_integrated_output_endpoint( continue_previous_state: bool, is_index: bool, ) -> Result, ControllerError> { - let registry = builtin_integrated_output_endpoint_registry(); - create_integrated_output_endpoint_with_registry( - ®istry, - endpoint_id, - endpoint_name, - connector_config, - key_schema, - schema, - controller, - continue_previous_state, - is_index, - ) -} - -#[allow(unused, clippy::too_many_arguments)] -pub fn create_integrated_output_endpoint_with_registry( - registry: &IntegratedOutputEndpointRegistry, - endpoint_id: EndpointId, - endpoint_name: &str, - connector_config: &ConnectorConfig, - key_schema: &Option, - schema: &Relation, - controller: Weak, - continue_previous_state: bool, - is_index: bool, -) -> Result, ControllerError> { - let ep = registry - .create_endpoint( + let ep: Box = match &connector_config.transport { + #[cfg(feature = "with-deltalake")] + TransportConfig::DeltaTableOutput(config) => Box::new(delta_table::DeltaTableWriter::new( endpoint_id, endpoint_name, - connector_config, + config, key_schema, schema, controller, continue_previous_state, is_index, - )? - .ok_or_else(|| { - ControllerError::unknown_output_transport( + )?), + TransportConfig::PostgresOutput(config) => Box::new(PostgresOutputEndpoint::new( + endpoint_id, + endpoint_name, + config, + key_schema, + schema, + controller, + is_index, + )?), + #[cfg(feature = "with-dynamodb")] + TransportConfig::DynamoDBOutput(config) => Box::new(DynamoDBOutputEndpoint::new( + endpoint_id, + endpoint_name, + config, + key_schema, + schema, + controller, + is_index, + )?), + transport => { + return Err(ControllerError::unknown_output_transport( endpoint_name, - &connector_config.transport.name(), - ) - })?; + &transport.name(), + )); + } + }; if connector_config.format.is_some() { return Err(ControllerError::invalid_parser_configuration( @@ -469,37 +113,43 @@ pub fn create_integrated_input_endpoint( runtime_env: Arc, consumer: Box, ) -> Result, ControllerError> { - let registry = builtin_integrated_input_endpoint_registry(); - create_integrated_input_endpoint_with_registry( - ®istry, - endpoint_name, - config, - pipeline_config, - runtime_env, - consumer, - ) -} - -#[allow(unused_variables)] -pub fn create_integrated_input_endpoint_with_registry( - registry: &IntegratedInputEndpointRegistry, - endpoint_name: &str, - config: &ConnectorConfig, - pipeline_config: &PipelineConfig, - runtime_env: Arc, - consumer: Box, -) -> Result, ControllerError> { - let ep = registry - .create_endpoint( + let ep: Box = match &config.transport { + #[cfg(feature = "with-deltalake")] + TransportConfig::DeltaTableInput(config) => { + Box::new(delta_table::DeltaTableInputEndpoint::new( + endpoint_name, + config, + 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)) + } + #[cfg(feature = "with-postgres-cdc")] + TransportConfig::PostgresCdcInput(config) => Box::new(PostgresCdcInputEndpoint::new( endpoint_name, config, - pipeline_config, - runtime_env, consumer, - )? - .ok_or_else(|| { - ControllerError::unknown_input_transport(endpoint_name, &config.transport.name()) - })?; + )), + transport => { + return Err(ControllerError::unknown_input_transport( + endpoint_name, + &transport.name(), + )); + } + }; if config.format.is_some() { return Err(ControllerError::invalid_parser_configuration( diff --git a/crates/adapters/src/transport.rs b/crates/adapters/src/transport.rs index d5ad42d7b21..61048cf3dee 100644 --- a/crates/adapters/src/transport.rs +++ b/crates/adapters/src/transport.rs @@ -82,33 +82,33 @@ pub use feldera_adapterlib::transport::*; pub fn builtin_input_transport_registry() -> InputTransportRegistry { let mut registry = InputTransportRegistry::new(); - registry.register("file_input", Box::new(FileInputFactory)); + registry.register(TransportConfig::FILE_INPUT, Box::new(FileInputFactory)); #[cfg(feature = "with-kafka")] - registry.register("kafka_input", Box::new(KafkaInputFactory)); + registry.register(TransportConfig::KAFKA_INPUT, Box::new(KafkaInputFactory)); #[cfg(feature = "with-nats")] - registry.register("nats_input", Box::new(NatsInputFactory)); + registry.register(TransportConfig::NATS_INPUT, Box::new(NatsInputFactory)); #[cfg(feature = "with-pubsub")] - registry.register("pub_sub_input", Box::new(PubSubInputFactory)); - registry.register("url_input", Box::new(UrlInputFactory)); - registry.register("s3_input", Box::new(S3InputFactory)); - registry.register("datagen", Box::new(DatagenInputFactory)); + 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("nexmark", Box::new(NexmarkInputFactory)); - registry.register("http_input", Box::new(HttpInputFactory)); - registry.register("adhoc_input", Box::new(AdHocInputFactory)); - registry.register("clock", Box::new(ClockInputFactory)); - registry.register("empty_input", Box::new(EmptyInputFactory)); + 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("file_output", Box::new(FileOutputFactory)); + registry.register(TransportConfig::FILE_OUTPUT, Box::new(FileOutputFactory)); #[cfg(feature = "with-kafka")] - registry.register("kafka_output", Box::new(KafkaOutputFactory)); + registry.register(TransportConfig::KAFKA_OUTPUT, Box::new(KafkaOutputFactory)); #[cfg(feature = "with-redis")] - registry.register("redis_output", Box::new(RedisOutputFactory)); - registry.register("null_output", Box::new(NullOutputFactory)); + registry.register(TransportConfig::REDIS_OUTPUT, Box::new(RedisOutputFactory)); + registry.register(TransportConfig::NULL_OUTPUT, Box::new(NullOutputFactory)); registry } @@ -488,7 +488,7 @@ mod tests { .unwrap() .is_none() ); - input_registry.register("empty_input", Box::new(EmptyInputFactory)); + input_registry.register(TransportConfig::EMPTY_INPUT, Box::new(EmptyInputFactory)); assert!( input_registry .create_endpoint(&TransportConfig::EmptyInput) @@ -503,7 +503,7 @@ mod tests { .unwrap() .is_none() ); - output_registry.register("null_output", Box::new(NullOutputFactory)); + output_registry.register(TransportConfig::NULL_OUTPUT, Box::new(NullOutputFactory)); assert!( output_registry .create_endpoint(&TransportConfig::NullOutput, "null", true) @@ -511,4 +511,33 @@ mod tests { .is_some() ); } + + #[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")] + 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")] + assert!(output_registry.get(TransportConfig::REDIS_OUTPUT).is_some()); + assert!(output_registry.get(TransportConfig::NULL_OUTPUT).is_some()); + } } diff --git a/crates/feldera-types/src/config.rs b/crates/feldera-types/src/config.rs index ad34a38e74d..e5ff58c3687 100644 --- a/crates/feldera-types/src/config.rs +++ b/crates/feldera-types/src/config.rs @@ -1861,32 +1861,57 @@ pub enum TransportConfig { } impl TransportConfig { + 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 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(), + TransportConfig::FileInput(_) => Self::FILE_INPUT.to_string(), + TransportConfig::FileOutput(_) => Self::FILE_OUTPUT.to_string(), + TransportConfig::NatsInput(_) => Self::NATS_INPUT.to_string(), + TransportConfig::KafkaInput(_) => Self::KAFKA_INPUT.to_string(), + TransportConfig::KafkaOutput(_) => Self::KAFKA_OUTPUT.to_string(), + TransportConfig::PubSubInput(_) => Self::PUB_SUB_INPUT.to_string(), + TransportConfig::UrlInput(_) => Self::URL_INPUT.to_string(), + TransportConfig::S3Input(_) => Self::S3_INPUT.to_string(), + TransportConfig::DeltaTableInput(_) => Self::DELTA_TABLE_INPUT.to_string(), + TransportConfig::DeltaTableOutput(_) => Self::DELTA_TABLE_OUTPUT.to_string(), + TransportConfig::DynamoDBOutput(_) => Self::DYNAMODB_OUTPUT.to_string(), + TransportConfig::IcebergInput(_) => Self::ICEBERG_INPUT.to_string(), + TransportConfig::PostgresInput(_) => Self::POSTGRES_INPUT.to_string(), + TransportConfig::PostgresCdcInput(_) => Self::POSTGRES_CDC_INPUT.to_string(), + TransportConfig::PostgresOutput(_) => Self::POSTGRES_OUTPUT.to_string(), + TransportConfig::Datagen(_) => Self::DATAGEN.to_string(), + TransportConfig::Nexmark(_) => Self::NEXMARK.to_string(), + TransportConfig::HttpInput(_) => Self::HTTP_INPUT.to_string(), + TransportConfig::HttpOutput(_) => Self::HTTP_OUTPUT.to_string(), + TransportConfig::AdHocInput(_) => Self::ADHOC_INPUT.to_string(), + TransportConfig::RedisOutput(_) => Self::REDIS_OUTPUT.to_string(), + TransportConfig::ClockInput(_) => Self::CLOCK.to_string(), + TransportConfig::NullOutput => Self::NULL_OUTPUT.to_string(), + TransportConfig::EmptyInput => Self::EMPTY_INPUT.to_string(), } } From 98ca127172c40dbad44b2de8884d351e3fec310a Mon Sep 17 00:00:00 2001 From: Dnreikronos Date: Tue, 23 Jun 2026 13:17:57 -0300 Subject: [PATCH 3/6] Make transport configs factory parsed refs #6446 --- crates/adapterlib/src/transport.rs | 16 +- crates/adapters/src/adhoc/table.rs | 2 +- crates/adapters/src/controller.rs | 33 ++- crates/adapters/src/format/avro/output.rs | 20 +- crates/adapters/src/format/csv.rs | 5 +- crates/adapters/src/format/json/output.rs | 17 +- crates/adapters/src/format/parquet.rs | 5 +- crates/adapters/src/integrated.rs | 81 +++--- crates/adapters/src/server.rs | 39 ++- crates/adapters/src/test/datagen.rs | 15 +- crates/adapters/src/transport.rs | 156 ++++------ crates/adapters/src/transport/clock.rs | 32 ++- .../adapters/src/transport/kafka/ft/test.rs | 51 ++-- crates/adapters/src/transport/s3.rs | 9 +- crates/feldera-types/src/config.rs | 170 ++++++----- crates/feldera-types/src/secret_resolver.rs | 24 +- .../src/compiler/sql_compiler.rs | 5 +- .../pipeline-manager/src/db/types/program.rs | 266 +++++++++++++++--- 18 files changed, 564 insertions(+), 382 deletions(-) diff --git a/crates/adapterlib/src/transport.rs b/crates/adapterlib/src/transport.rs index 7225408b055..5574fc35ac7 100644 --- a/crates/adapterlib/src/transport.rs +++ b/crates/adapterlib/src/transport.rs @@ -86,7 +86,7 @@ pub trait InputTransportEndpointFactory: Send + Sync { /// Registry of input transport endpoint factories keyed by transport name. #[derive(Default)] pub struct InputTransportRegistry { - registered: BTreeMap<&'static str, Arc>, + registered: BTreeMap>, } impl InputTransportRegistry { @@ -98,10 +98,10 @@ impl InputTransportRegistry { pub fn register( &mut self, - name: &'static str, + name: impl Into, factory: Box, ) { - self.registered.insert(name, Arc::from(factory)); + self.registered.insert(name.into(), Arc::from(factory)); } pub fn get(&self, name: &str) -> Option> { @@ -112,7 +112,7 @@ impl InputTransportRegistry { &self, config: &TransportConfig, ) -> AnyResult>> { - let Some(factory) = self.get(&config.name()) else { + let Some(factory) = self.get(config.name()) else { return Ok(None); }; factory.create(config) @@ -1151,7 +1151,7 @@ pub trait OutputTransportEndpointFactory: Send + Sync { /// Registry of output transport endpoint factories keyed by transport name. #[derive(Default)] pub struct OutputTransportRegistry { - registered: BTreeMap<&'static str, Arc>, + registered: BTreeMap>, } impl OutputTransportRegistry { @@ -1163,10 +1163,10 @@ impl OutputTransportRegistry { pub fn register( &mut self, - name: &'static str, + name: impl Into, factory: Box, ) { - self.registered.insert(name, Arc::from(factory)); + self.registered.insert(name.into(), Arc::from(factory)); } pub fn get(&self, name: &str) -> Option> { @@ -1179,7 +1179,7 @@ impl OutputTransportRegistry { endpoint_name: &str, fault_tolerant: bool, ) -> AnyResult>> { - let Some(factory) = self.get(&config.name()) else { + let Some(factory) = self.get(config.name()) else { return Ok(None); }; factory.create(config, endpoint_name, fault_tolerant) 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/controller.rs b/crates/adapters/src/controller.rs index 8864a11540f..d5e53da4470 100644 --- a/crates/adapters/src/controller.rs +++ b/crates/adapters/src/controller.rs @@ -4842,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 { @@ -6279,7 +6276,7 @@ impl ControllerInner { .input_transport_registry() .lock() .unwrap() - .get(&transport_config.name()); + .get(transport_config.name()); let endpoint = match factory { Some(factory) => factory .create(&transport_config) @@ -6412,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", @@ -6610,7 +6609,7 @@ impl ControllerInner { .output_transport_registry() .lock() .unwrap() - .get(&transport_config.name()); + .get(transport_config.name()); let endpoint = match factory { Some(factory) => factory .create( 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..e4e9cbc7373 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, @@ -53,41 +66,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 +126,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 61048cf3dee..d888d4c7990 100644 --- a/crates/adapters/src/transport.rs +++ b/crates/adapters/src/transport.rs @@ -119,12 +119,8 @@ impl InputTransportEndpointFactory for FileInputFactory { &self, config: &TransportConfig, ) -> AnyResult>> { - match config { - TransportConfig::FileInput(config) => { - Ok(Some(Box::new(FileInputEndpoint::new(config)))) - } - _ => Ok(None), - } + let config = config.deserialize_config()?; + Ok(Some(Box::new(FileInputEndpoint::new(config)))) } } @@ -137,12 +133,8 @@ impl InputTransportEndpointFactory for KafkaInputFactory { &self, config: &TransportConfig, ) -> AnyResult>> { - match config { - TransportConfig::KafkaInput(config) => { - Ok(Some(Box::new(KafkaFtInputEndpoint::new(config)?))) - } - _ => Ok(None), - } + let config = config.deserialize_config()?; + Ok(Some(Box::new(KafkaFtInputEndpoint::new(config)?))) } } @@ -155,12 +147,8 @@ impl InputTransportEndpointFactory for NatsInputFactory { &self, config: &TransportConfig, ) -> AnyResult>> { - match config { - TransportConfig::NatsInput(config) => { - Ok(Some(Box::new(NatsInputEndpoint::new(config)?))) - } - _ => Ok(None), - } + let config = config.deserialize_config()?; + Ok(Some(Box::new(NatsInputEndpoint::new(config)?))) } } @@ -173,12 +161,8 @@ impl InputTransportEndpointFactory for PubSubInputFactory { &self, config: &TransportConfig, ) -> AnyResult>> { - match config { - TransportConfig::PubSubInput(config) => { - Ok(Some(Box::new(PubSubInputEndpoint::new(config.clone())?))) - } - _ => Ok(None), - } + let config = config.deserialize_config()?; + Ok(Some(Box::new(PubSubInputEndpoint::new(config)?))) } } @@ -189,10 +173,8 @@ impl InputTransportEndpointFactory for UrlInputFactory { &self, config: &TransportConfig, ) -> AnyResult>> { - match config { - TransportConfig::UrlInput(config) => Ok(Some(Box::new(UrlInputEndpoint::new(config)))), - _ => Ok(None), - } + let config = config.deserialize_config()?; + Ok(Some(Box::new(UrlInputEndpoint::new(config)))) } } @@ -203,10 +185,8 @@ impl InputTransportEndpointFactory for S3InputFactory { &self, config: &TransportConfig, ) -> AnyResult>> { - match config { - TransportConfig::S3Input(config) => Ok(Some(Box::new(S3InputEndpoint::new(config)?))), - _ => Ok(None), - } + let config = config.deserialize_config()?; + Ok(Some(Box::new(S3InputEndpoint::new(config)?))) } } @@ -217,12 +197,8 @@ impl InputTransportEndpointFactory for DatagenInputFactory { &self, config: &TransportConfig, ) -> AnyResult>> { - match config { - TransportConfig::Datagen(config) => { - Ok(Some(Box::new(GeneratorEndpoint::new(config.clone())))) - } - _ => Ok(None), - } + let config = config.deserialize_config()?; + Ok(Some(Box::new(GeneratorEndpoint::new(config)))) } } @@ -235,12 +211,8 @@ impl InputTransportEndpointFactory for NexmarkInputFactory { &self, config: &TransportConfig, ) -> AnyResult>> { - match config { - TransportConfig::Nexmark(config) => { - Ok(Some(Box::new(NexmarkEndpoint::new(config.clone())))) - } - _ => Ok(None), - } + let config = config.deserialize_config()?; + Ok(Some(Box::new(NexmarkEndpoint::new(config)))) } } @@ -251,12 +223,8 @@ impl InputTransportEndpointFactory for HttpInputFactory { &self, config: &TransportConfig, ) -> AnyResult>> { - match config { - TransportConfig::HttpInput(config) => { - Ok(Some(Box::new(HttpInputEndpoint::new(config)))) - } - _ => Ok(None), - } + let config = config.deserialize_config()?; + Ok(Some(Box::new(HttpInputEndpoint::new(config)))) } } @@ -267,12 +235,8 @@ impl InputTransportEndpointFactory for AdHocInputFactory { &self, config: &TransportConfig, ) -> AnyResult>> { - match config { - TransportConfig::AdHocInput(config) => { - Ok(Some(Box::new(AdHocInputEndpoint::new(config)))) - } - _ => Ok(None), - } + let config = config.deserialize_config()?; + Ok(Some(Box::new(AdHocInputEndpoint::new(config)))) } } @@ -283,10 +247,8 @@ impl InputTransportEndpointFactory for ClockInputFactory { &self, config: &TransportConfig, ) -> AnyResult>> { - match config { - TransportConfig::ClockInput(config) => Ok(Some(Box::new(ClockEndpoint::new(config)?))), - _ => Ok(None), - } + let config = config.deserialize_config()?; + Ok(Some(Box::new(ClockEndpoint::new(config)?))) } } @@ -297,10 +259,7 @@ impl InputTransportEndpointFactory for EmptyInputFactory { &self, config: &TransportConfig, ) -> AnyResult>> { - match config { - TransportConfig::EmptyInput => Ok(Some(Box::new(EmptyInputEndpoint))), - _ => Ok(None), - } + Ok(Some(Box::new(EmptyInputEndpoint))) } } @@ -313,12 +272,8 @@ impl OutputTransportEndpointFactory for FileOutputFactory { _endpoint_name: &str, _fault_tolerant: bool, ) -> AnyResult>> { - match config { - TransportConfig::FileOutput(config) => { - Ok(Some(Box::new(FileOutputEndpoint::new(config)?))) - } - _ => Ok(None), - } + let config = config.deserialize_config()?; + Ok(Some(Box::new(FileOutputEndpoint::new(config)?))) } } @@ -333,15 +288,13 @@ impl OutputTransportEndpointFactory for KafkaOutputFactory { endpoint_name: &str, fault_tolerant: bool, ) -> AnyResult>> { - match config { - TransportConfig::KafkaOutput(config) => match fault_tolerant { - false => Ok(Some(Box::new(KafkaOutputEndpoint::new( - config, - endpoint_name, - )?))), - true => Ok(Some(Box::new(KafkaFtOutputEndpoint::new(config)?))), - }, - _ => Ok(None), + 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)?))), } } } @@ -357,12 +310,8 @@ impl OutputTransportEndpointFactory for RedisOutputFactory { _endpoint_name: &str, _fault_tolerant: bool, ) -> AnyResult>> { - match config { - TransportConfig::RedisOutput(config) => { - Ok(Some(Box::new(RedisOutputEndpoint::new(config)?))) - } - _ => Ok(None), - } + let config = config.deserialize_config()?; + Ok(Some(Box::new(RedisOutputEndpoint::new(config)?))) } } @@ -375,10 +324,7 @@ impl OutputTransportEndpointFactory for NullOutputFactory { _endpoint_name: &str, _fault_tolerant: bool, ) -> AnyResult>> { - match config { - TransportConfig::NullOutput => Ok(Some(Box::new(NullOutputEndpoint))), - _ => Ok(None), - } + Ok(Some(Box::new(NullOutputEndpoint))) } } @@ -427,7 +373,7 @@ mod tests { let secrets_dir = tempfile::tempdir().unwrap(); let endpoint = input_transport_config_to_endpoint( - &TransportConfig::EmptyInput, + &TransportConfig::without_config(TransportConfig::EMPTY_INPUT), "empty", secrets_dir.path(), ) @@ -442,7 +388,7 @@ mod tests { let secrets_dir = tempfile::tempdir().unwrap(); let endpoint = output_transport_config_to_endpoint( - &TransportConfig::NullOutput, + &TransportConfig::without_config(TransportConfig::NULL_OUTPUT), "null", true, secrets_dir.path(), @@ -460,7 +406,7 @@ mod tests { assert!( input_transport_config_to_endpoint( - &TransportConfig::NullOutput, + &TransportConfig::without_config(TransportConfig::NULL_OUTPUT), "null", secrets_dir.path() ) @@ -469,7 +415,7 @@ mod tests { ); assert!( output_transport_config_to_endpoint( - &TransportConfig::EmptyInput, + &TransportConfig::without_config(TransportConfig::EMPTY_INPUT), "empty", false, secrets_dir.path() @@ -484,32 +430,48 @@ mod tests { let mut input_registry = InputTransportRegistry::new(); assert!( input_registry - .create_endpoint(&TransportConfig::EmptyInput) + .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::EmptyInput) + .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::NullOutput, "null", true) + .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::NullOutput, "null", true) + .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] 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 e5ff58c3687..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,45 +1840,19 @@ 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 { @@ -1886,49 +1881,48 @@ impl TransportConfig { pub const NULL_OUTPUT: &'static str = "null_output"; pub const EMPTY_INPUT: &'static str = "empty_input"; - pub fn name(&self) -> String { - match self { - TransportConfig::FileInput(_) => Self::FILE_INPUT.to_string(), - TransportConfig::FileOutput(_) => Self::FILE_OUTPUT.to_string(), - TransportConfig::NatsInput(_) => Self::NATS_INPUT.to_string(), - TransportConfig::KafkaInput(_) => Self::KAFKA_INPUT.to_string(), - TransportConfig::KafkaOutput(_) => Self::KAFKA_OUTPUT.to_string(), - TransportConfig::PubSubInput(_) => Self::PUB_SUB_INPUT.to_string(), - TransportConfig::UrlInput(_) => Self::URL_INPUT.to_string(), - TransportConfig::S3Input(_) => Self::S3_INPUT.to_string(), - TransportConfig::DeltaTableInput(_) => Self::DELTA_TABLE_INPUT.to_string(), - TransportConfig::DeltaTableOutput(_) => Self::DELTA_TABLE_OUTPUT.to_string(), - TransportConfig::DynamoDBOutput(_) => Self::DYNAMODB_OUTPUT.to_string(), - TransportConfig::IcebergInput(_) => Self::ICEBERG_INPUT.to_string(), - TransportConfig::PostgresInput(_) => Self::POSTGRES_INPUT.to_string(), - TransportConfig::PostgresCdcInput(_) => Self::POSTGRES_CDC_INPUT.to_string(), - TransportConfig::PostgresOutput(_) => Self::POSTGRES_OUTPUT.to_string(), - TransportConfig::Datagen(_) => Self::DATAGEN.to_string(), - TransportConfig::Nexmark(_) => Self::NEXMARK.to_string(), - TransportConfig::HttpInput(_) => Self::HTTP_INPUT.to_string(), - TransportConfig::HttpOutput(_) => Self::HTTP_OUTPUT.to_string(), - TransportConfig::AdHocInput(_) => Self::ADHOC_INPUT.to_string(), - TransportConfig::RedisOutput(_) => Self::REDIS_OUTPUT.to_string(), - TransportConfig::ClockInput(_) => Self::CLOCK.to_string(), - TransportConfig::NullOutput => Self::NULL_OUTPUT.to_string(), - TransportConfig::EmptyInput => Self::EMPTY_INPUT.to_string(), + 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..061ae3ef792 100644 --- a/crates/pipeline-manager/src/db/types/program.rs +++ b/crates/pipeline-manager/src/db/types/program.rs @@ -697,6 +697,67 @@ impl ProgramInfo { } } +#[derive(Clone, Copy, 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, + } +} + /// 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 +780,16 @@ 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(); + let sql_direction = sql_connector_transport_direction(transport_name); + if sql_direction == Some(TransportDirection::Output) + || (known_transport_direction(transport_name).is_some() && sql_direction.is_none()) + { + 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 +824,16 @@ 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(); + let sql_direction = sql_connector_transport_direction(transport_name); + if sql_direction == Some(TransportDirection::Input) + || (known_transport_direction(transport_name).is_some() && sql_direction.is_none()) + { + 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,14 @@ 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, 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 +944,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 +1038,142 @@ 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], + }) + } + } + + #[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()); + } } From bb3e90af0459b257c5a870c38800d3e4675b34e7 Mon Sep 17 00:00:00 2001 From: Dnreikronos Date: Wed, 24 Jun 2026 11:29:37 -0300 Subject: [PATCH 4/6] Guard SQL transport direction drift --- .../pipeline-manager/src/db/types/program.rs | 76 ++++++++++++++++--- 1 file changed, 66 insertions(+), 10 deletions(-) diff --git a/crates/pipeline-manager/src/db/types/program.rs b/crates/pipeline-manager/src/db/types/program.rs index 061ae3ef792..9e4bbcefc7f 100644 --- a/crates/pipeline-manager/src/db/types/program.rs +++ b/crates/pipeline-manager/src/db/types/program.rs @@ -697,7 +697,7 @@ impl ProgramInfo { } } -#[derive(Clone, Copy, Eq, PartialEq)] +#[derive(Clone, Copy, Debug, Eq, PartialEq)] enum TransportDirection { Input, Output, @@ -758,6 +758,12 @@ fn sql_connector_transport_direction(name: &str) -> Option { } } +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( @@ -781,10 +787,7 @@ pub fn generate_program_info( .clone() .expect("Origin value cannot be None if connectors is non-empty"); let transport_name = connector.config.transport.name(); - let sql_direction = sql_connector_transport_direction(transport_name); - if sql_direction == Some(TransportDirection::Output) - || (known_transport_direction(transport_name).is_some() && sql_direction.is_none()) - { + if rejects_sql_connector_transport(transport_name, TransportDirection::Input) { return Err(ConnectorGenerationError::ExpectedInputConnector { position: origin_value.value_position, relation: input_relation.name.sql_name(), @@ -825,10 +828,7 @@ pub fn generate_program_info( .clone() .expect("Origin value cannot be None if connectors is non-empty"); let transport_name = connector.config.transport.name(); - let sql_direction = sql_connector_transport_direction(transport_name); - if sql_direction == Some(TransportDirection::Input) - || (known_transport_direction(transport_name).is_some() && sql_direction.is_none()) - { + if rejects_sql_connector_transport(transport_name, TransportDirection::Output) { return Err(ConnectorGenerationError::ExpectedOutputConnector { position: origin_value.value_position, relation: output_relation.name.sql_name(), @@ -905,7 +905,10 @@ pub fn generate_pipeline_config( #[cfg(test)] mod tests { - use super::{determine_connector_endpoint_names, generate_program_info, RuntimeSelector}; + 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, }; @@ -1081,6 +1084,59 @@ mod tests { } } + 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!( From 7921c6e78eb2667c8c61e33317270b488cf31ad3 Mon Sep 17 00:00:00 2001 From: Dnreikronos Date: Wed, 24 Jun 2026 11:30:01 -0300 Subject: [PATCH 5/6] Test transport secret resolution before parsing --- crates/adapters/src/transport.rs | 125 ++++++++++++++++++++++++++++++- 1 file changed, 123 insertions(+), 2 deletions(-) diff --git a/crates/adapters/src/transport.rs b/crates/adapters/src/transport.rs index d888d4c7990..ae67eea4801 100644 --- a/crates/adapters/src/transport.rs +++ b/crates/adapters/src/transport.rs @@ -338,9 +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)?; - builtin_input_transport_registry().create_endpoint(&config) + registry.create_endpoint(&config) } /// Creates an output transport endpoint instance using an output transport @@ -358,15 +368,69 @@ 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)?; - builtin_output_transport_registry().create_endpoint(&config, endpoint_name, fault_tolerant) + 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() { @@ -474,6 +538,63 @@ mod tests { 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(); From 8f99d9d62972f78d866b93d3a3ca1049e1408283 Mon Sep 17 00:00:00 2001 From: Dnreikronos Date: Wed, 24 Jun 2026 11:30:19 -0300 Subject: [PATCH 6/6] Document integrated transport registry boundary --- crates/adapters/src/integrated.rs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/crates/adapters/src/integrated.rs b/crates/adapters/src/integrated.rs index e4e9cbc7373..a5a2f48b651 100644 --- a/crates/adapters/src/integrated.rs +++ b/crates/adapters/src/integrated.rs @@ -55,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,