Writing a connector

dataferry has no plugin loading. A connector is a Rust crate, and you ship it by building your own binary: the dataferry crate is a library as well as a command, and a three-line main gives your binary every subcommand, the built-in connectors, and yours. The built-in postgres and mongo connectors are registered exactly this way, through the same public calls.

The design is decision 0004. This chapter walks through it with code from the repository that its tests compile and run.

The pieces

A connector crate needs only the first two (plus serde and schemars for its options, and usually async-trait), so it doesn’t pull in the planner or DataFusion.

The Connector trait

A connector declares its type: name, the roles it provides, and an options struct for its fields in connections.yaml, and builds endpoints from them. The repository’s README example, which a test compiles and registers:

use dataferry_registry::{BuildContext, BuildError, Connector, Endpoints, Roles, Secret};
use schemars::JsonSchema;
use serde::Deserialize;

pub struct ClickHouse;

/// A ClickHouse database. Doc comments become editor hover text.
#[derive(Deserialize, JsonSchema)]
#[serde(deny_unknown_fields)]
pub struct ClickHouseOptions {
    /// `clickhouse://` URL. A `Secret` is redacted from everything printed.
    url: Secret,
    /// The database written to.
    database: String,
}

impl Connector for ClickHouse {
    const TYPE: &'static str = "clickhouse";
    const ROLES: Roles = Roles::BOTH;
    type Options = ClickHouseOptions;

    fn build(
        &self,
        ctx: &BuildContext<'_>,
        options: ClickHouseOptions,
        wants: Roles,
    ) -> Result<Endpoints, BuildError> {
        // No I/O here: `dataferry validate` calls this. Keep the URL for the
        // endpoints, which connect on their first trait call.
        let ClickHouseOptions { url, database } = options;
        let _ = (ctx, url, database);
        todo!("the Source and/or Sink that {wants} asks for")
    }

    /// One line for `dataferry plan` and the run report. No secrets.
    fn describe(&self, options: &ClickHouseOptions) -> String {
        format!("clickhouse, database {}", options.database)
    }
}

• TYPE is the type: value. ROLES says whether it can be a source, a sink or both (Roles::SOURCE, Roles::SINK, Roles::BOTH): validate refuses a flow that uses a connection in another role.
• Options is your connection fields, minus name and type. Derive Deserialize with #[serde(deny_unknown_fields)], so a typo is an error, and JsonSchema, so dataferry schema describes it: doc comments become editor hover text, and defaults and enums carry through. Don’t use the reserved keys name, type or description. Options are validated by deserializing them: the Rust type is the source of truth, and the schema is derived from it.
• Secret for every credential-bearing field. It prints as ***, its schema is marked x-ferry-secret, validate warns when it holds a literal instead of ${ENV}, and its value is redacted from everything the CLI prints. Secrets inside another value (a password in a URL) are returned by derived_secrets (it defaults to none) and redacted too.
• build returns the Source and/or Sink that wants asks for (always a subset of ROLES), in an Endpoints. It must not do I/O: validate calls it to catch bad options without connecting. Endpoints connect on their first trait call. Report bad options as BuildError::option("<field>", "<message>"); the CLI turns the field into a file, line and column.
• describe is the one line plan and validate -v show for a connection. Never put a secret in it.

Your binary

fn main() -> std::process::ExitCode {
    dataferry::Ferry::with_builtins()
        .register(ClickHouse)
        .name("acme-dataferry")
        .main()
}

Ferry is a builder:

Registering a type twice, or a connector whose options use a reserved key, is a bug in the binary: every command then fails at start-up with exit 1 and a diagnostic (F0002) naming the type.

The subcommands and these builder methods are the library’s whole stable surface in this release.

The acme test connector

crates/dataferry/tests/acme-dataferry is a complete third-party binary, written against the public API only and used by the tests: it registers its own postgres and mongo (with the built-ins’ options, and endpoints that panic when used, which proves validate never connects) and a fake type the stock binary doesn’t have.

Its command line:

/// The binary's command line: what `main` runs, and what tests build on.
pub fn cli() -> Ferry {
    Ferry::new()
        .register_state_capable(Postgres, Backend)
        .register(Mongo)
        .register(Fake)
        .name("acme-dataferry")
}

A source-only connector, with a URL check in build that stands in for parsing a connection string, and no I/O:

// ---- mongo: a source only ----

pub struct Mongo;

/// A MongoDB database.
#[derive(Deserialize, JsonSchema)]
#[serde(deny_unknown_fields)]
pub struct MongoOptions {
    /// `mongodb://` or `mongodb+srv://`.
    url: Secret,
    database: String,
}

impl Connector for Mongo {
    const TYPE: &'static str = "mongo";
    const ROLES: Roles = Roles::SOURCE;
    type Options = MongoOptions;

    fn build(
        &self,
        ctx: &BuildContext<'_>,
        options: MongoOptions,
        wants: Roles,
    ) -> std::result::Result<Endpoints, BuildError> {
        check_url(&options.url, &["mongodb", "mongodb+srv"])?;
        Ok(endpoints(ctx, wants))
    }

    fn describe(&self, options: &MongoOptions) -> String {
        format!("mongo database {}", options.database)
    }
}

endpoints builds whatever wants asks for. Endpoints also has two optional fields for sinks: a finalizer (restores what prepare loosened for the load, such as dropped foreign keys) and ddl (renders the DDL prepare would run, which plan prints):

fn endpoints(ctx: &BuildContext<'_>, wants: Roles) -> Endpoints {
    let endpoint = Arc::new(Unreachable {
        connection: ctx.connection.to_owned(),
    });
    Endpoints {
        source: wants.source.then(|| endpoint.clone() as Arc<dyn Source>),
        sink: wants.sink.then_some(endpoint as Arc<dyn Sink>),
        finalizer: None,
        ddl: None,
    }
}

Its tests run the binary as a real process (validate against the repository’s example configs and its bad-config set, with every secret checked to never appear in the output), and in-process through run_with (registration mistakes, and the JSON Schema its registered types compose).

Implementing the source and sink

The traits are in dataferry-core. All are async_trait, Send + Sync, and the runner holds them as Arc<dyn Source> / Arc<dyn Sink>.

Source:

• discover() returns the entities: name, fields with canonical types and nullability, and the primary key. For a schemaless store, infer the schema; records that don’t fit it are surfaced at read time as rejected rows (dataferry_registry::rejected), not errors.
• plan_partitions(entity, hint) splits an entity into partitions that together cover every row exactly once, sized by hint.target_rows. Called once per run; the result is stored, never re-planned. When the hint carries a co-partition (a merge’s joined entity), produce exactly its ranges on its column, or return CoPartitionUnsupported (decision 0002).
• read(entity, partition) streams the partition’s rows as Arrow batches matching the entity’s schema. Every source must also accept a Whole partition.
• capabilities() says what the source can do (range scans, predicate pushdown, snapshots).

Sink:

• prepare(target_schema, mode) creates or checks the targets, once per flow.
• begin(partition, lease, strategy) opens a SinkTxn for one partition: write(target, batch) any number of times, then commit() once, or abort(). Everything a partition writes commits together.
• capabilities() lists the idempotency strategies the sink supports. The planner takes the first one compatible with the flow’s mode, in the order TxnWithState, DeletePartitionThenInsert, UpsertByKey, StagingSwap; a flow with none is refused. Whatever you offer must make re-running a partition safe (decision 0001).

Exactly-once: state-capable connectors

TxnWithState commits a partition’s rows and its done-record in one transaction. A connector offers it by implementing StateCapable (a pool() that connects lazily, and build_with_ledger, which builds a sink that calls the given Ledger on its own transaction just before COMMIT), paired with a StateBackend that keeps the state store in the same database, and is registered with register_state_capable. The compiler checks that the connector’s and the backend’s pool and connection types are the same. A sink built with plain build must not list TxnWithState. The acme connector’s version, which does nothing:

/// The driver's pool, connecting lazily: there is nothing to connect to.
#[derive(Clone)]
pub struct Pool;

impl StateCapable for Postgres {
    type Pool = Pool;
    type Conn = ();

    fn pool(
        &self,
        _: &BuildContext<'_>,
        _: &PostgresOptions,
    ) -> std::result::Result<Pool, BuildError> {
        Ok(Pool)
    }

    fn build_with_ledger(
        &self,
        ctx: &BuildContext<'_>,
        options: PostgresOptions,
        wants: Roles,
        _: Pool,
        _: Arc<dyn Ledger<()>>,
    ) -> std::result::Result<Endpoints, BuildError> {
        self.build(ctx, options, wants)
    }
}

/// A state store that panics on first use, like the endpoints.
pub struct Backend;

impl StateBackend for Backend {
    type Pool = Pool;
    type Conn = ();

    fn open(&self, _: Pool) -> (Arc<dyn StateStore>, Arc<dyn Ledger<()>>) {
        (Arc::new(Store), Arc::new(Store))
    }
}

Testing a connector

With the testkit features:

The contract suites return every violation they find, so a test asserts the list is empty; certify returns a certificate, or the first reason it refused one. The built-in connectors pass all four.

What a third-party connector can’t do yet

• Be verified. After a run, dataferry reads a sink back to verify it only for the types it knows how to read: postgres. A third-party sink’s flows are reported as not verified (W0701), which doesn’t change the exit code; what its finalizer couldn’t restore still fails the run.
• Qualify entity names with a namespace, as app.public.orders does for Postgres: that is also registered only for the built-ins.
• Be loaded at run time. Connectors are compiled in. A language-agnostic subprocess protocol is designed to fit later (decision 0004 §7), not built.

Using dataferry as a library

Beyond Ferry, the command functions (plan, run, …) are private in this release: embedding dataferry means running its commands through Ferry::run_with, with your own arguments and output buffers, and reading the exit code and the run report. The crates below the CLI (the DSL, the runner, the state store, the verifier) are published and documented on docs.rs, but their APIs are not stable yet.