v1.5.0

⚡ Highlights

🧮 Transformation rules work on array columns. Now each element is transformed, with an optional array_options to control the output
🔐 New fpe_ff1 transformer: format-preserving encryption for phone numbers, emails and codes, uniqueness preserved
🔎 New lookup_choice transformer: replaces a value with one read from a live table column
🔑 Kafka SASL authentication (plain, scram-sha-256, scram-sha-512) on source and target
⚡ Parallel index restore in pg-to-pg schema snapshots via index_restore_workers, and transient pg_restore failures are retried instead of aborting the snapshot
🧩 Transformers accept more column types: numeric in the greenmask number transformers, user-defined enums in greenmask_choice, any type in template

What's Changed

New features

• Add an array transformer that maps over array elements by @kvch in #1182
• Apply transformation rules to array columns by @kvch in #1184
• Add fpe_ff1 format-preserving transformer for emails, phone numbers by @kvch in #1165
• Add lookup_choice transformer by @kvch in #1148
• Add sasl to kafka by @RemiDesgrange in #1169
• Restore indexes in parallel during pg2pg schema snapshot by @kvch in #1144
• Retry index restore on transient errors by @kvch in #1139
• Support enum columns in greenmask_choice by @kvch in #1192
• Support numeric columns in transformers by @kvch in #1140
• Allow any column type in template transformer by @kvch in #1185
• Add context to transformer validation errors by @kvch in #1176

Bug fixes

• Stop retrying data exceptions and name the column by @kvch in #1175
• Bind the env vars the configuration reads by @kvch in #1145
• Encode snapshot values with pgx text codecs in templates by @kvch in #1188
• Parse range literals for binary COPY in bulk ingest by @kvch in #1187
• Decode array elements with a catalog aware type map by @kvch in #1189
• Warn and fail early on array column rule surprises by @kvch in #1190
• Resolve enum types and return choice labels as strings by @kvch in #1191

Dependency updates

• Bump github.com/moby/go-archive from 0.2.0 to 0.3.0 by @dependabot[bot] in #1171
• Bump go.opentelemetry.io/contrib/instrumentation/runtime from 0.70.0 to 0.71.0 by @dependabot[bot] in #1154
• Bump go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc from 1.45.0 to 1.46.0 by @dependabot[bot] in #1155
• Bump clickhouse/clickhouse-server from 26.7.5.10-alpine to 26.8.2.7-alpine by @dependabot[bot] in #1150

Details

Transformation rules on array columns

A rule on an array column such as varchar(500)[] was rejected at validation time with does not support pg data type: _varchar with OID: 1015. Because the rejection happened for the whole rule set, a single array column blocked transformation of every other column in the table. A rule on a one-dimensional array column now applies the transformer to each element.

transformations:
  table_transformers:
    - schema: public
      table: users
      column_transformers:
        emails: # varchar(500)[]
          name: email
          parameters:
            replacement_domain: "@example.com.invalid"

pgstream decodes the array before it calls the transformer, so each element arrives as the value of its element type: an int4[] element is an integer, not the text "1", and a NULL element is distinguishable from the string "NULL". A transformer accepts an array column if it accepts the element type, which extends to extension types (citext[]) and to arrays of enums. The array is re-encoded in the same shape it arrived in, so Kafka, webhook and search payloads for array columns are unchanged.

The optional array_options block controls how the output array is built:

column_transformers:
  emails:
    name: email
    array_options:
      generator: random # map (default) or random
      min_count: 0
      max_count: 12

map applies the transformer to each element in order and keeps the length. random emits between min_count and max_count elements, each the transform of a source element drawn at random with replacement, so it never invents a value that was not in the row. min_count and max_count are required with random, rejected with map, and max_count is capped at 10000. Array columns need a source PostgreSQL connection, since the element type is read from the catalog; the connection-less parser rejects array_options.

The uniqueness check reads the classification of the whole array transform: with map the array keeps the transformer's classification, so fpe_ff1 on text[] stays preserved; with random the array is always lossy, since min_count: 0 lets any row produce an empty array. A NULL array stays NULL, a NULL element stays NULL, and the first failing element fails the whole column under the configured on_error policy. Multi-dimensional arrays are not supported and are rejected at startup when declared as such.

Three things the validator now flags: literal_string and pg_anonymizer write into every element on an array column, which is a change for a configuration written before array support; pg_anonymizer sends one query per element, so a wide array multiplies the source load; and a column declared with more than one dimension is rejected up front instead of failing row by row. The first two are warnings in pgstream validate rules and in the run log.

fpe_ff1: format-preserving encryption

The output has the same length and the same characters as the input. An encrypted phone number is still a phone number, an encrypted product code still fits its varchar(n) column, and FF1 (NIST SP 800-38G) maps each input to one different output, so the transformer is classified preserved and can be used on a unique-indexed column. It is the first uniqueness-preserving transformer whose output keeps the original format.

transformations:
  table_transformers:
    - schema: public
      table: persons
      column_transformers:
        phone_number:
          name: fpe_ff1
          parameters:
            key_hex: ${FPE_KEY} # 32, 48 or 64 hex chars; openssl rand -hex 32
            associated_data: public.persons.phone_number # FF1 tweak, not secret
            alphabet: digits # digits, letters, alphanumeric, or a set of characters
            passthrough: keep # copy characters outside the alphabet to the output
            keep_prefix: 4 # keep the country and operator code
        email:
          name: fpe_ff1
          parameters:
            key_hex: ${FPE_KEY}
            associated_data: public.persons.email
            alphabet: alphanumeric
            passthrough: keep
            preserve_from: "." # keep the top-level domain

keep_prefix and keep_suffix count only characters in the alphabet, so one rule serves every phone format: +36301234567, +36 30 123 4567 and +36-30-123-4567 all keep their first four digits and their separators. preserve_from keeps the end of the value from the last occurrence of a delimiter: "." keeps the top-level domain so the address stays well-formed, "@" keeps the full domain but then the target contains every source domain. FF1 refuses domains smaller than 10⁶, so a value needs at least 6 digits, or 4 letters or alphanumerics, after prefix, suffix and passthrough characters are removed; shorter values go to the on_error policy rather than reaching the target in clear.

Like encrypted_aes_siv, this is pseudonymization: anyone holding key_hex can decrypt, and equal inputs give equal outputs. Keep the key in a secret store, use a different key per environment, and set associated_data per column so tokens cannot be correlated across columns.

lookup_choice: choose from a live table

greenmask_choice and greenmask_integer force users to regenerate their rules file whenever a lookup table changes. lookup_choice replaces a value with one taken from a column of another table, read from the source database once at pipeline start:

transformations:
  table_transformers:
    - schema: public
      table: addresses
      column_transformers:
        country_id:
          name: lookup_choice
          parameters:
            lookup_table: public.countries # schema-qualified, quoted as written
            lookup_column: id
            generator: deterministic # or random
            ignore_values: [0, -1] # drop placeholder rows from the pool
            # max_values: 100000   # load fails above this, rather than truncating
            # postgres_url: ...    # only needed when the source is not PostgreSQL

The type is taken from the lookup column, and a rule that points a column at an incompatible lookup type is rejected at startup; a narrower integer or float is accepted for a wider column. deterministic picks by hash of the incoming value so rows that shared an original still share a replacement, random picks per row. Deterministic mode is rejected for timestamp and timestamptz lookup columns, because snapshot and replication deliver them in forms that do not hash the same.

Please note that the transformer is lossy: any table with more rows than the lookup column has values produces duplicates, so it cannot be used on a unique-indexed column. The values come from the source lookup table, so if that table's key column is itself transformed, or a lookup row is deleted after start, the chosen value does not exist on the target and the foreign key rejects it; under replication that row is dropped with a DATALOSS line unless strict_mode is set. And the whole column is loaded into memory once per rule, capped at max_values, with a 30 second read timeout.

Kafka SASL

Both the Kafka source and the Kafka target now authenticate with SASL plain, scram-sha-256 or scram-sha-512. The mechanism is required when SASL is enabled rather than defaulting to plain, so a configuration that omits it fails. GSSAPI and OAUTHBEARER are not supported.

source:
  kafka:
    servers: ["localhost:9092"]
    topic:
      name: "mytopic"
    consumer_group:
      id: "mygroup"
    tls:
      ca_cert: "/path/to/ca.crt"
    sasl: # remove this section to disable SASL authentication
      mechanism: "scram-sha-512" # plain, scram-sha-256 or scram-sha-512. Required
      user: "myuser"
      password: "mypassword"

The same block goes under target.kafka. The environment variables are PGSTREAM_KAFKA_SASL_ENABLED, PGSTREAM_KAFKA_SASL_MECHANISM, PGSTREAM_KAFKA_SASL_USER and PGSTREAM_KAFKA_SASL_PASSWORD. plain sends the password in clear text, so enable TLS with it. Thanks to @RemiDesgrange for this contribution.

Parallel index restore

Index restore during a pg-to-pg schema snapshot ran as a single sequential psql session, so a schema with many indexes built them one at a time even though CREATE INDEX statements do not depend on each other. The new index_restore_workers setting restores standalone CREATE INDEX/CREATE UNIQUE INDEX statements concurrently. Everything that can depend on an index existing (constraints added USING INDEX, comments, partition attachments, REPLICA IDENTITY, CLUSTER) is restored afterwards, once every index has been created, in the original order.

source:
  postgres:
    snapshot:
      schema:
        mode: pgdump_pgrestore
        pgdump_pgrestore:
          index_constraint_session_settings:
            - maintenance_work_mem=4GB
          index_restore_workers: 8 # defaults to 1 (sequential), maximum 32

Each worker holds one target connection and applies index_constraint_session_settings to it, so the target needs that many free connection slots and up to workers × maintenance_work_mem of memory. A failing index does not cancel the restores still in flight; the per-index failures are merged and the retry decision is taken over the whole wave. Measured locally: 300 small indexes went from 4.8s to 1.7s with 8 workers, and 8 indexes on a 3M-row table from 34.4s to 11.0s with 4. The environment variable is PGSTREAM_POSTGRES_SNAPSHOT_INDEX_RESTORE_WORKERS.

Index restore retries transient failures

Any pg_restore failure during the index and constraint phase that was not recognised as already-exists, does-not-exist, permission or constraint became critical and aborted the whole snapshot. A dropped connection, a server still coming up or a lock that could not be taken in time therefore lost a snapshot that a second attempt would have completed. Restore output is now classified against a list of transient fragments (dropped or refused connections, SSL/EOF/broken pipe, server starting up or in recovery, connection-slot exhaustion, lock timeout, deadlock, serialization failure) and the index and constraint restore is retried with exponential backoff: 1s initial, 1min max, 3 retries.

More column types in transformers

Three additions:

• numeric columns are accepted by greenmask_float, greenmask_integer and greenmask_unix_timestamp, which previously failed with does not support pg data type: numeric with OID: 1700. For greenmask_float on a numeric column min_value and max_value are required, since the default range covers all float32 values and would write the same constant into every row. The range is checked against the column's numeric(p,s) at validation time. The source value is converted to float64 for the seed, so digits beyond float64 precision are rounded; this affects only the seed, since the transformer discards the source value.

• User-defined enum columns are accepted by greenmask_choice. choices becomes optional on an enum column: pgstream reads the labels from the catalog at startup and logs them, and an explicit choices list is checked against the labels so a typo stops the run rather than every insert. A domain over an enum, and an array of an enum, resolve to the enum. This needs a source PostgreSQL URL; a Kafka-sourced pipeline still has to set choices. Prefer generator: random on an enum: deterministic maps each label to one fixed label with no secret key, and pgstream warns when it sees that combination.

  column_transformers:
    mood: # a user-defined enum column
      name: greenmask_choice # no choices needed; the enum labels are used

• The template transformer declared only string and byte_array as compatible, so pgstream validate rejected it on an int4 column although the documentation said it works on any type. It now declares all, like literal_string and pg_anonymizer. That exposed a second gap: a snapshot row handed the template pgx Go types where replication handed it Postgres text, so {{ .GetValue }} rendered a date as 2024-02-29 00:00:00 +0000 UTC on a snapshot and the target rejected it. Snapshot values are now encoded with pgx's text codecs before the template runs, so a numeric renders as decimal text, a uuid as a0eebc99-…, a date as 2024-02-29, a bytea as \xdeadbeef, and one template gives the same output from either source. Values read with .GetDynamicValue keep their Go time.Time type so the sprig date functions still work on them.

Two related bulk-ingest fixes: a transformer that turns a range column into its literal ([1,10) from template or literal_string) failed binary COPY with cannot find encode plan; the literal is now scanned into the typed range pgx can encode, for int4range, int8range, numrange, daterange, tsrange and tstzrange. And the array transformer built its own type map that knew only built-in OIDs, so a rule on citext[] or a domain over an array passed validation and then nulled 100% of rows at run time under the default on_error: null, one DATALOSS line each, while the run reported success. It now uses the catalog-aware map from the connection, which also installs the raw JSON codec so jsonb[] elements are no longer round-tripped through float64.

Data exceptions stop the run and name the column

A snapshot with transformation rules could stall forever, logging data exception: value too long for type character varying(50) with a growing retry delay and never completing or erroring. The message named only the type, so it read as if a TEXT column were limited to 50 characters. Two causes: MapError kept PgError.Message and dropped the CONTEXT line (COPY ledger_line, line 1, column code) that names the table and column; and class 22 errors were not in the permanent set, so a failure raised by the row values themselves was replayed identically under a retry policy with no attempt bound. The context is now appended to the error, and ErrDataException is no longer retried.

Rule position in validation errors

pgstream validate rules reported a failing transformer without saying which rule produced it. Errors now carry the rule index, table and column:

table_transformers[2]: column 'name' in table "public"."test": template_transformer: error parsing template: template: :5: function "literal_string" not defined

New Contributors

• @RemiDesgrange made their first contribution in #1169

Full Changelog: v1.4.2...v1.5.0