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.

OAuth tokens

The root oauth package handles client credentials and refresh-token grants. It refreshes on demand and exposes Token(ctx) to Go drivers. HTTP manifests use the same implementation. Browser authorization and callback handlers are outside this package. Bind a run-scoped store to an existing connection refresh-token secret. The provider must implement filament.SecretRotator. PostgreSQL, SQLite, AWS and GCP providers support it.

import (
    "github.com/galaxy-io/filament/oauth"
    oauthsecrets "github.com/galaxy-io/filament/oauth/secrets"
)

store, err := oauthsecrets.FromRefreshToken(secretProvider, tenant, refreshSecretRef, binding)
if err != nil {
    return err
}

tokens, err := oauth.NewSource(oauth.Config{
    TokenURL:     tokenURL,
    ClientID:     clientID,
    ClientSecret: clientSecret,
    ClientAuth:   oauth.ClientSecretBasic,
    GrantType:    oauth.RefreshToken,
}, store)
if err != nil {
    return err
}

// Call before an API request. The source reuses unexpired tokens.
token, err := tokens.Token(ctx)
if err != nil {
    return err
}
token.SetAuthHeader(req)

A driver can depend on the small oauth.TokenSource interface. For an HTTP source, call source.SetOAuthStore(store) before Configure or TestConnection. This is an explicit Go integration hook. A driver can implement oauth/secrets.Configurable to identify its saved refresh-token field and authorization settings. The runner binds that secret through oauth/secrets.Bind before source configuration. Binding performs no secret reads or exchanges. Unsaved connection validation must not exchange rotating tokens. The store loads the refresh token on first use and keeps access tokens and expiry in memory for that run. Only a changed refresh token is written back. PostgreSQL and SQLite use an encrypted update of the existing secret row. AWS uses PutSecretValue. GCP uses AddSecretVersion and destroys the previous version. Rotation never creates a secret or updates the connection row. Refreshes serialize within the store. Separate runs have independent token state. No database transaction spans a token request and no coordination table is required. Token requests default to a 30-second timeout. Use oauth.NewMemoryStore for temporary credentials or tests. The environment provider is read-only and cannot persist rotating tokens. Other integrations can implement oauth.Store or use oauth.StoreFunc. A returned token is available only after any refresh-token write succeeds. A failed save retains the exchange in memory for another save attempt without exchanging the consumed token again. A process exit between upstream rotation and persistence can require fresh authorization. Check errors.Is(err, oauth.ErrReauthorizationRequired) for an expired or revoked refresh grant. Endpoint errors expose HTTP status and an allowlisted OAuth code, without provider descriptions or raw responses. Token requests do not retry automatically. The HTTP adapter’s optional retry policy accepts explicit temporary rejections and avoids retrying ambiguous transport failures. Use the connection’s existing secret lifecycle when removing an authorization. Provider-side revocation remains a separate operation. Token values must never appear in logs, events, ordinary connection config or API responses.

Get started

Guides

Concepts

Usage

SDKs

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

OAuth tokens

​Start with one full-read source

​Resources and parts

​Row metadata for recovery and CDC

​Describe and register the source

​Add capabilities deliberately

​OAuth tokens

Start with one full-read source

Resources and parts

Row metadata for recovery and CDC

Describe and register the source

Add capabilities deliberately

OAuth tokens