streamling advanced plugin kinds — Agent Skill

Beyond sources, transforms, and sinks, a plugin crate can register three more component kinds — preprocessors, UDFs, and side outputs — and can mix all six in one cdylib. This skill also covers the low-level manual FFI escape hatch for when the registration macros don't fit.

Prerequisite: [streamling-plugin-basics](skill://streamling-plugin-basics).

Preprocessors — rewrite topology YAML before parsing

A preprocessor transforms the raw topology config string (the pipeline YAML) before streamling parses it. Use it to inject default options, expand shorthand dataset names into concrete source configs, or enforce per-source conventions — without touching streamling's core config parser.

use async_trait::async_trait;
use std::collections::HashMap;
use streamling_plugin::api::PluginError;
use streamling_plugin::{PluginInitializationError, PreprocessorPlugin};

pub struct DefaultsPreprocessor {
    default_region: String,
}

impl DefaultsPreprocessor {
    /// Constructor takes ONLY options (no schema/rt/state/metrics).
    pub fn new(options: HashMap<String, String>) -> Result<Self, PluginInitializationError> {
        Ok(Self {
            default_region: options.get("region").cloned().unwrap_or_else(|| "us-east-1".into()),
        })
    }
}

#[async_trait]
impl PreprocessorPlugin for DefaultsPreprocessor {
    async fn preprocess_topology(&self, config: String) -> Result<String, PluginError> {
        // Parse, mutate, re-serialize. serde_yaml is the usual tool.
        let mut doc: serde_yaml::Value = serde_yaml::from_str(&config)
            .map_err(|e| PluginError::Internal(format!("parse topology: {e}")))?;

        if let Some(sinks) = doc.get_mut("sinks").and_then(|v| v.as_mapping_mut()) {
            for (_, sink) in sinks.iter_mut() {
                if let Some(map) = sink.as_mapping_mut() {
                    let key = serde_yaml::Value::String("region".into());
                    map.entry(key).or_insert_with(|| serde_yaml::Value::String(self.default_region.clone()));
                }
            }
        }

        serde_yaml::to_string(&doc).map_err(|e| PluginError::Internal(format!("emit topology: {e}")))
    }
}

Register and order matters — preprocessors run in registration order, and later ones see earlier ones' output:

register_plugin_preprocessor!("my_plugin", "defaults", DefaultsPreprocessor);

Discipline:

• Never overwrite a user-set field. Check existence first (entry().orinsert(..) / containskey) and only fill gaps.
• Preprocessors are async but should stay fast and deterministic — they run once at pipeline load.
• Parse failures must become PluginError::Internal, never panics.

UDFs — custom functions usable in SQL

A UDF is a DataFusion scalar function your plugin exposes so pipeline SQL can call it (SELECT myfunc(col) FROM ...) — the lightest way to add a derived column. It implements ScalarUDFImpl and registers with registerpluginudf!(MyFunc) (or registerpluginudffn!(createfn) for the legacy createudf closure form).

Full coverage — the modern invokewithargs / ScalarFunctionArgs API, a complete worked example, argument helpers, returntype vs returnfieldfromargs, volatility, and the legacy path — lives in its own skill: [streamling-udf-plugin](skill://streamling-udf-plugin).

Side outputs — observe every source

A side output observes data from all sources without joining the pipeline or modifying records — e.g. a monitor that logs batch counts, tracks a high-water mark, or emits a heartbeat. Unlike other kinds, side outputs use direct FFI invocation (no channels), and one instance is created per source (the macro manages a HashMap<source_name, instance>).

use arrow::array::RecordBatch;
use streamling_plugin::SideOutputPlugin;

pub struct MonitorSideOutput;

impl SideOutputPlugin for MonitorSideOutput {
    fn process_batch(&self, batch: &RecordBatch) -> Result<(), String> {
        tracing::info!(rows = batch.num_rows(), "observed source batch");
        Ok(())
    }
    fn shutdown(&self) { tracing::info!("side output shutting down"); }
}

Register with an id:

register_plugin_side_output!("monitor", MonitorSideOutput);

Note: SideOutputPlugin is sync and returns Result<(), String> (a plain String error, not PluginError). Instances are auto-registered against every source; you don't wire a from:.

Registering many kinds in one crate

A single cdylib can register sources, transforms, sinks, preprocessors, UDFs, and side outputs together — this is how production plugin crates are structured. Group them by mod, then list every macro call above the single initpluginwithasyncruntime!():

mod sinks; mod sources; mod transforms; mod udfs; mod preprocessors;

use crate::preprocessors::DefaultsPreprocessor;
use crate::sinks::{MySink, QueueSink};
use crate::sources::ApiSource;
use crate::transforms::FilterTransform;
use crate::udfs::BytesToHexFunc;

use streamling_plugin::{
    init_plugin_with_async_runtime, register_plugin_preprocessor, register_plugin_sink,
    register_plugin_source, register_plugin_transform, register_plugin_udf,
    set_plugin_output_buffer,
};

register_plugin_source!("my_plugin", "api_source", ApiSource);
set_plugin_output_buffer!("my_plugin.api_source", 1);

register_plugin_transform!("my_plugin", "filter", FilterTransform);
set_plugin_output_buffer!("my_plugin.filter", 1);

register_plugin_sink!("my_sink", MySink);
register_plugin_sink!("queue_sink", QueueSink);

register_plugin_preprocessor!("defaults", DefaultsPreprocessor);
register_plugin_udf!(BytesToHexFunc);

init_plugin_with_async_runtime!();

Rules:

• Exactly one initpluginwithasyncruntime!(), at the bottom, regardless of how many components you registered.
• setpluginoutputbuffer!/setplugininputbuffer! are optional, per-component, and take the full plugin.id.
• Keep each component in its own module/file; only the registration list lives in lib.rs.

Low-level manual FFI (escape hatch)

When the registration macros can't express what you need (a component constructed outside the standard closure, custom channel wiring), drop to the raw PluginModule ABI. This is advanced — prefer the macros whenever they fit. Shape (from the low_level example):

mod sink;

use abi_stable::std_types::{RNone, ROption, RResult, RString, RVec};
use abi_stable::{export_root_module, prefix_type::PrefixTypeTrait};
use streamling_plugin::api::PluginStateBackendFactory;
use streamling_plugin::r#async::PluginAsyncRuntimeObj;
use streamling_plugin::ffi::SafeArrowSchema;
use streamling_plugin::{
    PluginChannels, PluginInitializationError, PluginLogging, PluginModule, PluginModuleRef,
    PluginOptions, PluginResult, PluginRuntimeConfiguration, PluginUdfDescriptor,
    PluginSideOutputDescriptor, RootModule, sink_generator,
};
use streamling_plugin::IntoSinkPluginResult;

use crate::sink::MySink;

extern "C" fn init(logging: PluginLogging) -> RResult<PluginRuntimeConfiguration, PluginInitializationError> {
    logging.init_logger();
    RResult::ROk(PluginRuntimeConfiguration { /* ... */ })
}

extern "C" fn create(
    plugin_id: RString,
    input_schema: ROption<SafeArrowSchema>,
    options: PluginOptions,
    runtime: PluginAsyncRuntimeObj,
    state_backend_config: streamling_plugin::PluginStateBackendConfig,
    message_channels: PluginChannels,
) -> RResult<PluginResult, PluginInitializationError> {
    let schema: arrow_schema::SchemaRef = input_schema.into_rust().unwrap().into();  // for sinks/transforms
    sink_generator(
        plugin_id,
        |schema, rt, state, metrics, opts| {
            IntoSinkPluginResult::into_sink_result(MySink::new(schema, rt, state, metrics, opts))
        },
        options,
        runtime,
        state_backend_config,
        message_channels,
    )
}

extern "C" fn udf_descriptors() -> RResult<RVec<PluginUdfDescriptor>, PluginInitializationError> {
    RResult::ROk(RVec::new())
}
extern "C" fn side_output_descriptors() -> RResult<RVec<PluginSideOutputDescriptor>, PluginInitializationError> {
    RResult::ROk(RVec::new())
}

#[export_root_module]
pub fn get_module() -> PluginModuleRef {
    PluginModule::new(init, create, udf_descriptors, side_output_descriptors)
}

Use the low-level path only when you must hand-route create or channel handling; otherwise the macros generate this boilerplate correctly and keep you ABI-safe.

Common mistakes

• Preprocessor overwrites a user field. Always check before writing; users must be able to override your defaults.
• UDF invoke panics on bad arity/types. Validate and return a DataFusionError.
• Side output returns PluginError. It returns Result<(), String> — different signature from the other kinds.
• Two initpluginwithasyncruntime!() calls, or none. Exactly one per crate.
• Reaching for low-level FFI when a macro works. The macros handle ABI-stability details you can easily get wrong by hand.

Related skills

• [streamling-plugin-basics](skill://streamling-plugin-basics) — crate setup, registration, lifecycle, options/secrets, errors.
• [streamling-source-plugin](skill://streamling-source-plugin) / [streamling-sink-plugin](skill://streamling-sink-plugin) / [streamling-transform-plugin](skill://streamling-transform-plugin) — the main three kinds.
• [streamling-udf-plugin](skill://streamling-udf-plugin) — custom SQL functions via DataFusion ScalarUDFImpl.
• [sql-transform](skill://sql-transform) — SQL transform, the no-code alternative to a transform plugin or a UDF.