---
name: streamling-advanced-plugins
description: Use when implementing a streamling preprocessor (rewriting pipeline topology YAML before it parses), a DataFusion scalar UDF usable in SQL, a side output (observing every source without joining the pipeline), a partitioned source/transform/sink (one instance per physical stream), registering many component kinds in one crate, or dropping below the registration macros to the low-level manual FFI plugin module. Assumes streamling-plugin-basics.
---

# 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`. Sources, transforms, and sinks can also be **partitioned**, running one instance per physical stream. 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.

```rust
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:
```rust
register_plugin_preprocessor!("my_plugin", "defaults", DefaultsPreprocessor);
```

**Discipline:**
- **Never overwrite a user-set field.** Check existence first (`entry().or_insert(..)` / `contains_key`) 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 my_func(col) FROM ...`) — the lightest way to add a derived column. It implements `ScalarUDFImpl` and registers with `register_plugin_udf!(MyFunc)` (or `register_plugin_udf_fn!(create_fn)` for the legacy `create_udf` closure form).

**Full coverage — the modern `invoke_with_args` / `ScalarFunctionArgs` API, a complete worked example, argument helpers, `return_type` vs `return_field_from_args`, 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>`).

```rust
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:
```rust
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:`.

## Partitioned plugins — one instance per physical stream

A source, transform, or sink registered with the `register_plugin_*!` macros is **single-stream**: the host runs one instance and narrows its input to one stream. Register it with `register_partitioned_plugin_source!` / `_transform!` / `_sink!` instead and the host runs **one instance per physical stream**, each with its own channels. Use this when shards of the stream can be processed independently (a source with native shards, a keyed sink or transform).

Implement `PartitionedSourcePlugin` / `PartitionedTransformPlugin` / `PartitionedSinkPlugin` on top of the usual trait:

- `describe(...)` runs at planning time, **without** constructing an instance: validate options and report the output schema (sources and transforms), labels, input placement (transforms and sinks), and a `PartitionCount { minimum, maximum, preferred }`. Don't open connections that outlive the call or reserve durable resources; read-only schema discovery is fine.
- `create(context, ...)` is called once per partition. `context` carries `reference_name`, `partition_index`, and `partition_count`. Every instance must produce the described schema and labels.

```rust
use streamling_plugin::{
    InputPlacement, PartitionCount, PartitionedSinkPlugin, PluginInitializationError,
    PluginInstanceContext, SinkDescription,
};

impl PartitionedSinkPlugin for ShardedSink {
    fn describe(
        _input_schema: SchemaRef,
        options: &HashMap<String, String>,
    ) -> Result<SinkDescription, PluginInitializationError> {
        Ok(SinkDescription {
            labels: vec![],
            input_placement: InputPlacement::ByPrimaryKey,
            partition_count: PartitionCount { maximum: Some(16), ..PartitionCount::default() },
        })
    }

    fn create(
        context: PluginInstanceContext,
        input_schema: SchemaRef,
        rt: PluginAsyncRuntimeObj,
        state: PluginStateBackendFactory,
        metrics: PluginMetricsRecorder,
        options: HashMap<String, String>,
    ) -> Result<Self, PluginInitializationError> {
        // connect to shard `context.partition_index` ...
    }
}

register_partitioned_plugin_sink!("my_plugin", "sharded_sink", ShardedSink);
```

How the host runs it:

- **Width.** `parallelism` on the node sets it. Without it, a source runs `preferred` partitions (else 1) and a transform or sink inherits its input's width. A width outside `minimum..=maximum` fails planning, and so does a sink planned at a width other than its own `parallelism`: sinks that read the same node share one exchange, which runs at the widest `parallelism` among them, or on a single stream if any of them is single-stream or they declare different primary keys.
- **Input placement.** Before routing stream `i` to instance `i`, the host places rows as declared: `ByPrimaryKey` (the node's `primary_key`; for a transform whose key names columns it generates, the upstream node's key), `ByColumns(..)`, `RoundRobin`, or `Forward`. `RoundRobin` means any instance will do: when the node's `parallelism` differs from its input's width, whole batches are dealt across instances, so an insert and the delete that undoes it can reach different instances in any order. `Forward` (transforms only) plans no exchange: instance `i` reads input stream `i` in order, the transform runs as wide as its input, and a `parallelism` on it fails planning. Use it for a stateless or per-stream transform over an already partitioned input.
- **State.** `state.create()` is scoped to the partition (`{reference_name}[{index}]`); `state.create_shared()` is shared by every instance of the node (`{reference_name}`, the key a single-stream plugin's `create()` uses). See *State: decide before converting* below.
- **Checkpoints.** Each instance sees one copy of every marker on its own stream. A sink instance acks from `process_checkpoint_marker` as usual; the host acks the epoch once every instance flushed it.
- **Failures.** An error from a hook is reported to the host right away, which fails the pipeline and drains it.
- **Compatibility.** A host that predates partitioned plugins creates a partition-aware plugin through the single-stream `create`, as partition 0 of 1: exactly what a partitioned host runs at width 1, with the same state keys. So a plugin can ship before the engine, and the engine can be rolled back under it. Such a host passes `parallelism` through as an option, so the plugin refuses one other than 1 there, and refuses to start if its `describe` asks for a `minimum` above 1; a `preferred` width above 1 only logs a warning. Keep `streamling-plugin` on the same 0.x minor: abi_stable treats a 0.x minor bump as breaking, and an old host then refuses the whole library. A single-stream plugin rejects `parallelism` above 1.

### State: decide before converting

A partition-aware plugin always gets a partition-scoped state factory, at every width and on an old host too:

| | single-stream plugin | partition-aware plugin, instance `i` |
|---|---|---|
| `state.create()` | `{reference_name}` | `{reference_name}[i]` |
| `state.create_shared()` | `{reference_name}` | `{reference_name}` |

⚠️ Converting a single-stream plugin therefore moves its `create()` state to `{reference_name}[0]`, even at width 1. Nothing fails: every running pipeline restarts with empty state (a source from its configured start). Decide per piece of state:

- **Node-wide, `create_shared()`.** Always an option: the key the single-stream plugin wrote, at every width, so existing pipelines keep their state and width changes need no migration. The instances must coordinate writes (one writer, a distinct `_kv` key per partition, or one value they agree on, like a low watermark across partitions written once per checkpoint). Every instance of a node runs in one process, so they can coordinate in memory.
- **Per partition, `create()`.** Never redistributed: when `parallelism` changes, an index keeps its key but may own a different slice, indices past the new width are orphaned, and new ones start empty. Use it only when the slice an index owns cannot change (one native shard per partition) or the state can be rebuilt.
- **Migrate once.** An instance that finds no `create()` state can seed it from the old state under `create_shared()`, if that state can be split between partitions.

The same table is in the `PluginStateBackendFactory` docs.

## 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** `init_plugin_with_async_runtime!()`:

```rust
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** `init_plugin_with_async_runtime!()`, at the bottom, regardless of how many components you registered.
- `set_plugin_output_buffer!`/`set_plugin_input_buffer!` 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):

```rust
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)
}
```

A hand-written module must also fill the fields added after these: `set_shutdown_signal` (install the host's signal with `streamling_plugin::shutdown::install_shutdown_signal`), `describe_partitioned` (return `RResult::ROk(RNone)` for single-stream plugin ids), and `create_partitioned` (only called for ids `describe_partitioned` described). `plugin_examples/low_level` shows all of them.

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 `init_plugin_with_async_runtime!()` 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.