Writing a source

A source owns the read from an upstream system. Filament owns what happens afterward: bounded batching, backpressure, integrity checks, sink writes, and run progress. The important boundary is typed and columnar. A source describes a resource’s schema, opens a row writer for that resource, and appends values in schema order. Filament turns those rows into Arrow batches.

Start with one full-read source

The base interface is intentionally small:

type Source interface {
    Spec() ConnectorSpec
    Validate(cfg Config) error
    Configure(ctx context.Context, cfg Config) error
    Extract(ctx context.Context, sink RecordSink, opts ExtractOpts) error
    Teardown(ctx context.Context) error
}

One source instance serves one run. Configure can therefore keep a client, connection pool, or parsed configuration on the struct, and Teardown releases it when the run ends. Extract is where data crosses into Filament. This minimal example writes two columns for every requested resource:

func (s *Source) Extract(
    ctx context.Context,
    sink filament.RecordSink,
    opts filament.ExtractOpts,
) error {
    for _, resource := range opts.Resources {
        schema := filament.RecordSchema{
            Resource:   resource,
            PrimaryKey: []string{"id"},
            Fields: []filament.SchemaField{
                {Name: "id", Logical: filament.LogicalInt64},
                {Name: "name", Logical: filament.LogicalString},
            },
        }

        writer, err := sink.Builder(resource, 0, schema)
        if err != nil {
            return err
        }

        for _, row := range []struct {
            id   int64
            name string
        }{{1, "Ada"}, {2, "Linus"}} {
            if err := ctx.Err(); err != nil {
                return err
            }
            writer.Int64(row.id)
            writer.String(row.name)
            if err := writer.EndRow(filament.RowMeta{}); err != nil {
                return err
            }
        }
    }
    return nil
}

Every row must append exactly one value per schema field, in the same order. Use Null() for a null value. EndRow closes the row and may block while a full batch waits for the sink; returning its error is how backpressure and pipeline failures reach the source.

The pipeline closes builders after extraction. A source normally calls neither Close nor Drain; those are lifecycle details for advanced resumable or CDC sources. A long-lived stream may call Flush when it goes idle and needs to send a partially filled batch.

Resources and parts

A builder belongs to one (resource, part) pair. Most sources start with part 0. A source that splits a large table into independently read ranges opens one builder per range and assigns each a stable part number. ExtractOpts.Parallelism is the run’s concurrency budget, not an instruction to create that many parts. The source decides whether its backend can safely read in parallel. Each builder must be written by only one goroutine.

Row metadata for recovery and CDC

Most full-read sources can pass an empty RowMeta. Recovery and CDC sources use it to carry information that travels with a row without becoming a destination column:

Op is insert, update, or delete. Full reads normally leave it at insert.
Key is the source position for a keyset-resumable part.
Coarse marks progress that can only be acknowledged by row count.
LSN and Seq order a CDC stream.

These fields drive ordering and checkpoints. They are not a substitute for the resource schema or primary key.

Describe and register the source

Spec supplies the catalog name, configuration fields, and supported read policies. Keep the advertisement aligned with behavior: advertising incremental or CDC support does not implement the corresponding extraction interfaces. Register a factory so each run receives a fresh instance:

func init() {
    registry.RegisterSource(
        "mysource",
        filament.MaturityAlpha,
        func() filament.Source { return New() },
    )
}

Consumers include the connector with a blank import. Start at alpha, then raise maturity when the connector has been exercised end to end.

Add capabilities deliberately

A useful next step is Discoverable, which lets users browse resources before creating a pipeline. A SchemaProvider is required when the source must feed a schema-aware sink. Resume, incremental reads, and CDC require several contracts to work together; they are not single-interface switches. See Optional interfaces for those capability sets. The repository’s connectors/sample package is the smallest complete source. It demonstrates the same builder flow without an external dependency.

Get started

Concepts

Usage

Connectors

Deployment

API reference

Benchmarks

Contributing

Start with one full-read source

Resources and parts

Row metadata for recovery and CDC

Describe and register the source

Add capabilities deliberately

​Start with one full-read source

​Resources and parts

​Row metadata for recovery and CDC

​Describe and register the source

​Add capabilities deliberately

Start with one full-read source

Resources and parts

Row metadata for recovery and CDC

Describe and register the source

Add capabilities deliberately